Skip to content

Commit b25edcd

Browse files
zero88zero88
committed
feat(#93): TriggerEvaluator#afterTrigger
1 parent 60b7cfc commit b25edcd

File tree

13 files changed

+234
-176
lines changed

13 files changed

+234
-176
lines changed

‎core/src/main/java/io/github/zero88/schedulerx/impl/AbstractScheduler.java

Lines changed: 42 additions & 37 deletions
Original file line numberDiff line numberDiff line change
@@ -160,8 +160,10 @@ protected final void doStart(WorkerExecutor workerExecutor) {
160160
}
161161

162162
protected final void doStop(long timerId, TriggerContext context) {
163-
unregisterTimer(timerId);
164-
onComplete(context);
163+
if (context.isStopped()) {
164+
unregisterTimer(timerId);
165+
onComplete(context);
166+
}
165167
}
166168

167169
/**
@@ -174,18 +176,6 @@ protected final void doStop(long timerId, TriggerContext context) {
174176
*/
175177
protected abstract void unregisterTimer(long timerId);
176178

177-
/**
178-
* Check a trigger context whether to be able to stop by configuration or force stop
179-
*/
180-
protected final TriggerContext shouldStop(@NotNull TriggerContext triggerContext, boolean isForceStop, long round) {
181-
if (isForceStop) {
182-
return TriggerContextFactory.stop(triggerContext, ReasonCode.STOP_BY_JOB);
183-
}
184-
return trigger().shouldStop(round)
185-
? TriggerContextFactory.stop(triggerContext, ReasonCode.STOP_BY_CONFIG)
186-
: triggerContext;
187-
}
188-
189179
/**
190180
* Register a timer id in internal state and increase tick time when the system timer fires
191181
*
@@ -199,29 +189,27 @@ protected final long onFire(long timerId) {
199189
/**
200190
* Processing the trigger right away after the system timer fires
201191
*/
202-
protected final void onProcess(WorkerExecutor workerExecutor, TriggerContext ctx) {
203-
log(Objects.requireNonNull(ctx.firedAt()), "On fire");
204-
final Duration timeout = timeoutPolicy().evaluationTimeout();
205-
this.<TriggerContext>executeBlocking(workerExecutor, p -> this.wrapTimeout(timeout, p)
206-
.handle(evaluator.beforeRun(trigger, ctx,
207-
jobData.externalId())))
208-
.onSuccess(context -> onTrigger(workerExecutor, context))
209-
.onFailure(t -> onMisfire(TriggerContextFactory.skip(ctx, t instanceof TimeoutException
210-
? ReasonCode.EVALUATION_TIMEOUT
211-
: ReasonCode.UNEXPECTED_ERROR, t)));
192+
protected final void onProcess(WorkerExecutor workerExecutor, TriggerContext triggerContext) {
193+
log(Objects.requireNonNull(triggerContext.firedAt()), "On fire");
194+
this.onEvaluationBeforeTrigger(workerExecutor, triggerContext)
195+
.onSuccess(ctx -> onTrigger(workerExecutor, ctx))
196+
.onFailure(t -> onMisfire(TriggerContextFactory.skip(triggerContext, t instanceof TimeoutException
197+
? ReasonCode.EVALUATION_TIMEOUT
198+
: ReasonCode.UNEXPECTED_ERROR, t)));
212199
}
213200

214201
protected final void onTrigger(WorkerExecutor workerExecutor, TriggerContext triggerContext) {
215202
if (!triggerContext.isReady()) {
216203
onMisfire(triggerContext);
217204
return;
218205
}
219-
final ExecutionContextInternal<OUT> executionContext = new ExecutionContextImpl<>(vertx, triggerContext,
220-
state.increaseRound());
206+
final long round = state.increaseRound();
207+
final ExecutionContextInternal<OUT> executionContext = new ExecutionContextImpl<>(vertx, triggerContext, round);
221208
final Duration timeout = timeoutPolicy().executionTimeout();
222-
log(executionContext.triggeredAt(), "On trigger", triggerContext.tick(), executionContext.round());
223-
this.executeBlocking(workerExecutor, p -> executeJob(executionContext.setup(wrapTimeout(timeout, p))))
224-
.onComplete(ar -> onResult(executionContext, ar.cause()));
209+
log(executionContext.triggeredAt(), "On trigger", triggerContext.tick(), round);
210+
Future.join(onEvaluationAfterTrigger(workerExecutor, triggerContext, round),
211+
executeBlocking(workerExecutor, p -> executeJob(executionContext.setup(wrapTimeout(timeout, p)))))
212+
.onComplete(ar -> onResult(executionContext, ar.result().cause(1)));
225213
}
226214

227215
protected final void onSchedule(long timerId) {
@@ -260,6 +248,25 @@ protected final void onUnableSchedule(Throwable cause) {
260248
.build());
261249
}
262250

251+
protected final Future<TriggerContext> onEvaluationBeforeTrigger(WorkerExecutor worker, TriggerContext ctx) {
252+
return executeBlocking(worker, p -> {
253+
log(Instant.now(), "On before trigger");
254+
this.wrapTimeout(timeoutPolicy().evaluationTimeout(), p)
255+
.handle(evaluator.beforeTrigger(trigger, ctx, jobData.externalId()));
256+
});
257+
}
258+
259+
protected final Future<TriggerContext> onEvaluationAfterTrigger(WorkerExecutor worker, TriggerContext ctx,
260+
long round) {
261+
return executeBlocking(worker, p -> {
262+
log(Instant.now(), "On after trigger");
263+
wrapTimeout(timeoutPolicy().evaluationTimeout(), p).handle(
264+
evaluator.afterTrigger(trigger(), ctx, jobData.externalId(), round)
265+
.onSuccess(_ctx -> doStop(state.timerId(), _ctx))
266+
.onFailure(t -> LOGGER.error(genMsg(ctx.tick(), round, Instant.now(), "After evaluate"), t)));
267+
});
268+
}
269+
263270
protected final void onMisfire(@NotNull TriggerContext triggerCtx) {
264271
final Instant finishedAt = state.markFinished(triggerCtx.tick());
265272
final String reasonCode = triggerCtx.condition().reasonCode();
@@ -288,7 +295,7 @@ protected final void onResult(@NotNull ExecutionContext<OUT> executionContext, @
288295
if (asyncCause instanceof TimeoutException) {
289296
LOGGER.warn(genMsg(triggerContext.tick(), ctx.round(), finishedAt, asyncCause.getMessage()));
290297
} else if (asyncCause != null) {
291-
LOGGER.error(genMsg(triggerContext.tick(), ctx.round(), finishedAt, "System error"), asyncCause);
298+
LOGGER.error(genMsg(triggerContext.tick(), ctx.round(), finishedAt, "On result::System error"), asyncCause);
292299
}
293300
monitor.onEach(ExecutionResultImpl.<OUT>builder()
294301
.setExternalId(jobData.externalId())
@@ -304,9 +311,8 @@ protected final void onResult(@NotNull ExecutionContext<OUT> executionContext, @
304311
.setError(state.addError(ctx.round(),
305312
Optional.ofNullable(ctx.error()).orElse(asyncCause)))
306313
.build());
307-
final TriggerContext transitionCtx = shouldStop(triggerContext, ctx.isForceStop(), ctx.round());
308-
if (transitionCtx.isStopped()) {
309-
doStop(state.timerId(), transitionCtx);
314+
if (ctx.isForceStop()) {
315+
doStop(state.timerId(), TriggerContextFactory.stop(triggerContext, ReasonCode.STOP_BY_JOB));
310316
}
311317
}
312318

@@ -358,15 +364,15 @@ private <R> Promise<R> wrapTimeout(Duration timeout, Promise<R> promise) {
358364
}
359365

360366
@SuppressWarnings("rawtypes")
361-
private static class InternalTriggerEvaluator extends AbstractTriggerEvaluator {
367+
private static class InternalTriggerEvaluator extends DefaultTriggerEvaluator {
362368

363369
private final AbstractScheduler scheduler;
364370

365371
private InternalTriggerEvaluator(AbstractScheduler scheduler) { this.scheduler = scheduler; }
366372

367373
@Override
368-
protected Future<TriggerContext> internalCheck(@NotNull Trigger trigger, @NotNull TriggerContext ctx,
369-
@Nullable Object externalId) {
374+
protected Future<TriggerContext> internalBeforeTrigger(@NotNull Trigger trigger, @NotNull TriggerContext ctx,
375+
@Nullable Object externalId) {
370376
if (!ctx.isKickoff()) {
371377
throw new IllegalStateException("Trigger condition status must be " + TriggerStatus.KICKOFF);
372378
}
@@ -375,7 +381,6 @@ protected Future<TriggerContext> internalCheck(@NotNull Trigger trigger, @NotNul
375381

376382
@NotNull
377383
private TriggerContext doCheck(TriggerContext ctx) {
378-
scheduler.log(Instant.now(), "On evaluate");
379384
if (scheduler.state.pending()) {
380385
return TriggerContextFactory.skip(ctx, ReasonCode.NOT_YET_SCHEDULED);
381386
}

‎core/src/main/java/io/github/zero88/schedulerx/impl/AbstractSchedulerBuilder.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -49,7 +49,7 @@ public abstract class AbstractSchedulerBuilder<IN, OUT, T extends Trigger, S ext
4949

5050
@Override
5151
public @NotNull TriggerEvaluator triggerEvaluator() {
52-
return Optional.ofNullable(evaluator).orElseGet(AbstractTriggerEvaluator::noop);
52+
return Optional.ofNullable(evaluator).orElseGet(DefaultTriggerEvaluator::noop);
5353
}
5454

5555
@Override

‎core/src/main/java/io/github/zero88/schedulerx/impl/AbstractTriggerEvaluator.java

Lines changed: 0 additions & 45 deletions
This file was deleted.

‎core/src/main/java/io/github/zero88/schedulerx/impl/DefaultTriggerEvaluator.java

Lines changed: 57 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,57 @@
1+
package io.github.zero88.schedulerx.impl;
2+
3+
import org.jetbrains.annotations.ApiStatus.Internal;
4+
import org.jetbrains.annotations.NotNull;
5+
import org.jetbrains.annotations.Nullable;
6+
7+
import io.github.zero88.schedulerx.trigger.Trigger;
8+
import io.github.zero88.schedulerx.trigger.TriggerContext;
9+
import io.github.zero88.schedulerx.trigger.TriggerEvaluator;
10+
import io.vertx.core.Future;
11+
12+
@Internal
13+
public class DefaultTriggerEvaluator implements TriggerEvaluator {
14+
15+
static TriggerEvaluator noop() {
16+
return new DefaultTriggerEvaluator();
17+
}
18+
19+
private TriggerEvaluator next;
20+
21+
@Override
22+
public final @NotNull Future<TriggerContext> beforeTrigger(@NotNull Trigger trigger,
23+
@NotNull TriggerContext triggerContext,
24+
@Nullable Object externalId) {
25+
return this.internalBeforeTrigger(trigger, triggerContext, externalId)
26+
.flatMap(c -> next == null ? Future.succeededFuture(c) : next.beforeTrigger(trigger, c, externalId));
27+
}
28+
29+
@Override
30+
public final @NotNull Future<TriggerContext> afterTrigger(@NotNull Trigger trigger,
31+
@NotNull TriggerContext triggerContext,
32+
@Nullable Object externalId, long round) {
33+
// @formatter:off
34+
return this.internalAfterTrigger(trigger, triggerContext, externalId, round )
35+
.flatMap(c -> next == null ? Future.succeededFuture(c) : next.afterTrigger(trigger, c, externalId, round));
36+
// @formatter:on
37+
}
38+
39+
@Override
40+
public final @NotNull TriggerEvaluator andThen(@Nullable TriggerEvaluator another) {
41+
this.next = another;
42+
return this;
43+
}
44+
45+
protected Future<TriggerContext> internalBeforeTrigger(@NotNull Trigger trigger,
46+
@NotNull TriggerContext triggerContext,
47+
@Nullable Object externalId) {
48+
return Future.succeededFuture(triggerContext);
49+
}
50+
51+
protected Future<TriggerContext> internalAfterTrigger(@NotNull Trigger trigger,
52+
@NotNull TriggerContext triggerContext,
53+
@Nullable Object externalId, long round) {
54+
return Future.succeededFuture(triggerContext);
55+
}
56+
57+
}

‎core/src/main/java/io/github/zero88/schedulerx/trigger/EventSchedulerImpl.java

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,7 @@
1414
import io.github.zero88.schedulerx.TimeoutPolicy;
1515
import io.github.zero88.schedulerx.impl.AbstractScheduler;
1616
import io.github.zero88.schedulerx.impl.AbstractSchedulerBuilder;
17-
import io.github.zero88.schedulerx.impl.AbstractTriggerEvaluator;
17+
import io.github.zero88.schedulerx.impl.DefaultTriggerEvaluator;
1818
import io.github.zero88.schedulerx.impl.TriggerContextFactory;
1919
import io.github.zero88.schedulerx.trigger.TriggerCondition.ReasonCode;
2020
import io.github.zero88.schedulerx.trigger.predicate.EventTriggerPredicate.EventTriggerPredicateException;
@@ -97,12 +97,12 @@ static final class EventSchedulerBuilderImpl<IN, OUT, T>
9797
}
9898

9999

100-
static final class EventTriggerEvaluator<T> extends AbstractTriggerEvaluator {
100+
static final class EventTriggerEvaluator<T> extends DefaultTriggerEvaluator {
101101

102102
@Override
103103
@SuppressWarnings("unchecked")
104-
protected Future<TriggerContext> internalCheck(@NotNull Trigger trigger, @NotNull TriggerContext ctx,
105-
@Nullable Object externalId) {
104+
protected Future<TriggerContext> internalBeforeTrigger(@NotNull Trigger trigger, @NotNull TriggerContext ctx,
105+
@Nullable Object externalId) {
106106
try {
107107
if (ctx.condition().status() == TriggerCondition.TriggerStatus.READY &&
108108
!((EventTrigger<T>) trigger).getPredicate().test((T) ctx.info())) {

‎core/src/main/java/io/github/zero88/schedulerx/trigger/IntervalSchedulerImpl.java

Lines changed: 20 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -5,14 +5,17 @@
55
import java.time.Instant;
66

77
import org.jetbrains.annotations.NotNull;
8+
import org.jetbrains.annotations.Nullable;
89

910
import io.github.zero88.schedulerx.Job;
1011
import io.github.zero88.schedulerx.JobData;
1112
import io.github.zero88.schedulerx.SchedulingMonitor;
1213
import io.github.zero88.schedulerx.TimeoutPolicy;
1314
import io.github.zero88.schedulerx.impl.AbstractScheduler;
1415
import io.github.zero88.schedulerx.impl.AbstractSchedulerBuilder;
16+
import io.github.zero88.schedulerx.impl.DefaultTriggerEvaluator;
1517
import io.github.zero88.schedulerx.impl.TriggerContextFactory;
18+
import io.github.zero88.schedulerx.trigger.TriggerCondition.ReasonCode;
1619
import io.vertx.core.Future;
1720
import io.vertx.core.Promise;
1821
import io.vertx.core.Vertx;
@@ -24,7 +27,7 @@ final class IntervalSchedulerImpl<IN, OUT> extends AbstractScheduler<IN, OUT, In
2427
IntervalSchedulerImpl(@NotNull Job<IN, OUT> job, @NotNull JobData<IN> jobData, @NotNull TimeoutPolicy timeoutPolicy,
2528
@NotNull SchedulingMonitor<OUT> monitor, @NotNull IntervalTrigger trigger,
2629
@NotNull TriggerEvaluator evaluator, @NotNull Vertx vertx) {
27-
super(job, jobData, timeoutPolicy, monitor, trigger, evaluator, vertx);
30+
super(job, jobData, timeoutPolicy, monitor, trigger, new IntervalTriggerEvaluator().andThen(evaluator), vertx);
2831
}
2932

3033
protected @NotNull Future<Long> registerTimer(WorkerExecutor workerExecutor) {
@@ -67,4 +70,20 @@ static final class IntervalSchedulerBuilderImpl<IN, OUT>
6770

6871
}
6972

73+
74+
static final class IntervalTriggerEvaluator extends DefaultTriggerEvaluator {
75+
76+
@Override
77+
protected Future<TriggerContext> internalAfterTrigger(@NotNull Trigger trigger,
78+
@NotNull TriggerContext triggerContext,
79+
@Nullable Object externalId, long round) {
80+
IntervalTrigger interval = (IntervalTrigger) trigger;
81+
if (interval.noRepeatIndefinitely() && round >= interval.getRepeat()) {
82+
return Future.succeededFuture(TriggerContextFactory.stop(triggerContext, ReasonCode.STOP_BY_CONFIG));
83+
}
84+
return Future.succeededFuture(triggerContext);
85+
}
86+
87+
}
88+
7089
}

‎core/src/main/java/io/github/zero88/schedulerx/trigger/IntervalTrigger.java

Lines changed: 0 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -80,11 +80,6 @@ default long delayInMilliseconds() {
8080
@Override
8181
@NotNull IntervalTrigger validate();
8282

83-
@Override
84-
default boolean shouldStop(long round) {
85-
return noRepeatIndefinitely() && round >= getRepeat();
86-
}
87-
8883
@Override
8984
default JsonObject toJson() {
9085
JsonObject self = JsonObject.of("repeat", getRepeat(), "initialDelay", getInitialDelay(),

‎core/src/main/java/io/github/zero88/schedulerx/trigger/Trigger.java

Lines changed: 0 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -44,17 +44,6 @@ public interface Trigger extends HasTriggerType, TriggerRepresentation {
4444
*/
4545
@NotNull Trigger validate();
4646

47-
/**
48-
* Verify the execution should be stopped after the current execution round is out of the trigger rule.
49-
* <p/>
50-
* This method will be invoked right away after each execution round is finished regardless of the execution result
51-
* is success or error.
52-
*
53-
* @param round the current execution round
54-
* @since 2.0.0
55-
*/
56-
default boolean shouldStop(long round) { return false; }
57-
5847
/**
5948
* Simulate the next trigger times based on default preview parameter({@link PreviewParameter#byDefault()})
6049
*

‎core/src/main/java/io/github/zero88/schedulerx/trigger/TriggerEvaluator.java

Lines changed: 15 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -13,15 +13,24 @@
1313
public interface TriggerEvaluator {
1414

1515
/**
16-
* Check whether the trigger is able to run
16+
* Verify if the trigger can run before each execution round is started.
1717
*
18-
* @param trigger the trigger
19-
* @param context the trigger context
20-
* @param externalId the job external id
18+
* @param trigger the trigger
19+
* @param triggerContext the trigger context
20+
* @param externalId the job external id
2121
* @return a future of the trigger context that is evaluated
2222
*/
23-
@NotNull Future<TriggerContext> beforeRun(@NotNull Trigger trigger, @NotNull TriggerContext context,
24-
@Nullable Object externalId);
23+
@NotNull Future<TriggerContext> beforeTrigger(@NotNull Trigger trigger, @NotNull TriggerContext triggerContext,
24+
@Nullable Object externalId);
25+
26+
/**
27+
* Verify if the trigger should stop executing immediately after one round of execution begins.
28+
*
29+
* @param round the current execution round
30+
* @since 2.0.0
31+
*/
32+
@NotNull Future<TriggerContext> afterTrigger(@NotNull Trigger trigger, @NotNull TriggerContext triggerContext,
33+
@Nullable Object externalId, long round);
2534

2635
/**
2736
* Chain another evaluator

0 commit comments

Comments
 (0)