feat(#60): Implement EventTrigger

Author: zero88

.github/workflows/ci.yml

@@ -43,6 +43,7 @@ jobs:
       version: ${{ needs.context.outputs.version }}
       semanticVersion: ${{ needs.context.outputs.semanticVersion }}
       hashVersion: ${{ needs.context.outputs.commitId }}
+      buildArgs: 'clean build -x test'
 
   docs:
     uses: zero88/shared-ghactions/.github/workflows/antora-docs.yml@main

core/src/main/java/io/github/zero88/schedulerx/TaskExecutionContext.java

@@ -5,6 +5,7 @@
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
 
+import io.github.zero88.schedulerx.trigger.TriggerContext;
 import io.vertx.core.Vertx;
 
 /**
@@ -43,6 +44,13 @@ public interface TaskExecutionContext<OUTPUT> {
      */
     @NotNull Instant executedAt();
 
+    /**
+     * Runtime trigger context
+     *
+     * @since 2.0.0
+     */
+    @NotNull TriggerContext triggerContext();
+
     /**
      * Check whether force stop execution or not
      *

core/src/main/java/io/github/zero88/schedulerx/TaskExecutor.java

@@ -1,8 +1,5 @@
 package io.github.zero88.schedulerx;
 
-import org.jetbrains.annotations.NotNull;
-
-import io.vertx.codegen.annotations.GenIgnore;
 import io.vertx.codegen.annotations.VertxGen;
 import io.vertx.core.WorkerExecutor;
 
@@ -29,14 +26,6 @@ public interface TaskExecutor<INPUT, OUTPUT> extends TaskExecutorProperties<INPU
      */
     void start(WorkerExecutor workerExecutor);
 
-    /**
-     * Execute task
-     *
-     * @param executionContext execution context
-     */
-    @GenIgnore(GenIgnore.PERMITTED_TYPE)
-    void executeTask(@NotNull TaskExecutionContext<OUTPUT> executionContext);
-
     /**
      * Cancel executor
      */

core/src/main/java/io/github/zero88/schedulerx/TriggerTaskExecutor.java

@@ -3,15 +3,18 @@
 import org.jetbrains.annotations.NotNull;
 
 import io.github.zero88.schedulerx.trigger.Trigger;
+import io.github.zero88.schedulerx.trigger.TriggerContext;
+import io.vertx.codegen.annotations.GenIgnore;
 
 /**
  * Represents for an executor run task based on particular trigger
  *
  * @param <INPUT>   Type of Input Job data
  * @param <OUTPUT>  Type of Result data
  * @param <TRIGGER> Type of Trigger
- * @see Trigger
  * @see TaskExecutor
+ * @see Trigger
+ * @see TriggerContext
  * @since 1.0.0
  */
 public interface TriggerTaskExecutor<INPUT, OUTPUT, TRIGGER extends Trigger> extends TaskExecutor<INPUT, OUTPUT> {
@@ -23,4 +26,12 @@ public interface TriggerTaskExecutor<INPUT, OUTPUT, TRIGGER extends Trigger> ext
      */
     @NotNull TRIGGER trigger();
 
+    /**
+     * Execute task
+     *
+     * @param executionContext execution context
+     */
+    @GenIgnore(GenIgnore.PERMITTED_TYPE)
+    void executeTask(TaskExecutionContext<OUTPUT> executionContext);
+
 }

core/src/main/java/io/github/zero88/schedulerx/impl/AbstractTaskExecutor.java

@@ -4,6 +4,7 @@
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 
+import org.jetbrains.annotations.ApiStatus.Internal;
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
 
@@ -15,6 +16,7 @@
 import io.github.zero88.schedulerx.TaskResult;
 import io.github.zero88.schedulerx.TriggerTaskExecutor;
 import io.github.zero88.schedulerx.trigger.Trigger;
+import io.github.zero88.schedulerx.trigger.TriggerContext;
 import io.vertx.core.AsyncResult;
 import io.vertx.core.Future;
 import io.vertx.core.Promise;
@@ -30,6 +32,7 @@
  * @param <OUT> Type of output data
  * @param <T>   Type of trigger
  */
+@Internal
 public abstract class AbstractTaskExecutor<IN, OUT, T extends Trigger> implements TriggerTaskExecutor<IN, OUT, T> {
 
     @SuppressWarnings("java:S3416")
@@ -112,7 +115,7 @@ public final void start(WorkerExecutor workerExecutor) {
     @Override
     public final void executeTask(@NotNull TaskExecutionContext<OUT> executionContext) {
         try {
-            trace(state.tick(), state.round(), executionContext.executedAt(), "Start to execute the task");
+            trace(executionContext.executedAt(), "Start to execute the task");
             task.execute(jobData(), executionContext);
             if (!task.isAsync()) {
                 ((TaskExecutionContextInternal<OUT>) executionContext).internalComplete();
@@ -125,59 +128,66 @@ public final void executeTask(@NotNull TaskExecutionContext<OUT> executionContex
     @Override
     public final void cancel() {
         if (!state.completed()) {
-            trace(state.tick(), state.round(), Instant.now(), "Canceling the task");
+            trace(Instant.now(), "Canceling the task");
             doStop(state.timerId());
-            onCompleted();
         }
     }
 
-    protected abstract @NotNull Future<Long> addTimer(@NotNull Promise<Long> promise, WorkerExecutor workerExecutor);
+    protected final void doStart(WorkerExecutor workerExecutor) {
+        this.registerTimer(Promise.promise(), workerExecutor)
+            .onSuccess(this::onSchedule)
+            .onFailure(this::onUnableSchedule);
+    }
+
+    protected final void doStop(long timerId) {
+        unregisterTimer(timerId);
+        onCompleted();
+    }
+
+    protected abstract @NotNull Future<Long> registerTimer(@NotNull Promise<Long> promise,
+                                                           @Nullable WorkerExecutor workerExecutor);
 
-    protected final boolean shouldRun(@NotNull Instant triggerAt) {
-        final long tick = state.increaseTick();
+    protected void unregisterTimer(long timerId) { vertx.cancelTimer(timerId); }
+
+    protected InternalTriggerContext shouldRun(@NotNull Instant triggerAt, @NotNull TriggerContext triggerContext) {
+        state.increaseTick();
         if (state.completed()) {
-            trace(tick, state.round(), triggerAt, "The task execution is already completed");
+            trace(triggerAt, "The task execution is already completed");
         }
         if (state.executing()) {
-            trace(tick, state.round(), triggerAt, "Skip the execution due to the task is still running");
-            monitor.onMisfire(TaskResultImpl.<OUT>builder()
-                                            .setExternalId(jobData.externalId())
-                                            .setAvailableAt(state.availableAt())
-                                            .setTick(state.tick())
-                                            .setTriggeredAt(triggerAt)
-                                            .build());
+            trace(triggerAt, "Skip the execution due to the task is still running");
+            onMisfire(triggerAt);
         }
-        return state.idle() && trigger().shouldExecute(triggerAt);
+        return InternalTriggerContext.create(state.idle() && trigger().shouldExecute(triggerAt), triggerContext);
     }
 
     protected final boolean shouldStop(@Nullable TaskExecutionContext<OUT> executionContext, long round) {
         return (executionContext != null && executionContext.isForceStop()) || trigger().shouldStop(round);
     }
 
-    protected final void doStart(WorkerExecutor workerExecutor) {
-        this.addTimer(Promise.promise(), workerExecutor)
-            .onSuccess(this::onReceiveTimer)
-            .onFailure(t -> monitor.onUnableSchedule(TaskResultImpl.<OUT>builder()
-                                                                   .setExternalId(jobData.externalId())
-                                                                   .setTick(state.tick())
-                                                                   .setRound(state.round())
-                                                                   .setAvailableAt(state.availableAt())
-                                                                   .setUnscheduledAt(Instant.now())
-                                                                   .setError(t)
-                                                                   .build()));
-    }
-
-    protected void doStop(long timerId) {
-        vertx.cancelTimer(timerId);
+    protected final void run(WorkerExecutor workerExecutor, TriggerContext triggerContext) {
+        final Instant triggerAt = Instant.now();
+        final InternalTriggerContext internalContext = shouldRun(triggerAt, triggerContext);
+        if (internalContext.shouldRun()) {
+            final TriggerContext triggerCtx = TriggerContext.create(internalContext.type(), internalContext.info());
+            final TaskExecutionContextInternal<OUT> ctx = new TaskExecutionContextImpl<>(vertx, state.increaseRound(),
+                                                                                         triggerAt, triggerCtx);
+            trace(triggerAt, "Trigger the task execution");
+            if (workerExecutor != null) {
+                workerExecutor.executeBlocking(promise -> executeTask(onExecute(promise, ctx)), this::onResult);
+            } else {
+                vertx.executeBlocking(promise -> executeTask(onExecute(promise, ctx)), this::onResult);
+            }
+        }
     }
 
-    protected final void trace(long tick, long round, @NotNull Instant at, @NotNull String event) {
+    protected final void trace(@NotNull Instant at, @NotNull String event) {
         if (LOGGER.isTraceEnabled()) {
-            LOGGER.trace(genMsg(tick, round, at, event));
+            LOGGER.trace(genMsg(state.tick(), state.round(), at, event));
         }
     }
 
-    protected final void onReceiveTimer(long timerId) {
+    protected final void onSchedule(long timerId) {
         TaskResult<OUT> result;
         if (state.pending()) {
             result = TaskResultImpl.<OUT>builder()
@@ -196,37 +206,44 @@ protected final void onReceiveTimer(long timerId) {
         monitor.onSchedule(result);
     }
 
-    protected final void run(WorkerExecutor workerExecutor) {
-        final Instant triggerAt = Instant.now();
-        if (shouldRun(triggerAt)) {
-            TaskExecutionContextInternal<OUT> ctx = new TaskExecutionContextImpl<>(vertx, state.increaseRound(),
-                                                                                   triggerAt);
-            trace(state.tick(), ctx.round(), triggerAt, "Trigger the task execution");
-            if (workerExecutor != null) {
-                workerExecutor.executeBlocking(promise -> executeTask(onExecute(promise, ctx)), this::onResult);
-            } else {
-                vertx.executeBlocking(promise -> executeTask(onExecute(promise, ctx)), this::onResult);
-            }
-        }
+    protected final void onUnableSchedule(Throwable t) {
+        monitor.onUnableSchedule(TaskResultImpl.<OUT>builder()
+                                               .setExternalId(jobData.externalId())
+                                               .setTick(state.tick())
+                                               .setRound(state.round())
+                                               .setUnscheduledAt(Instant.now())
+                                               .setError(t)
+                                               .build());
     }
 
-    private TaskExecutionContextInternal<OUT> onExecute(@NotNull Promise<Object> promise,
-                                                        @NotNull TaskExecutionContextInternal<OUT> executionContext) {
-        state.markExecuting();
-        return executionContext.setup(promise, Instant.now());
+    protected final void onMisfire(@NotNull Instant triggerAt) {
+        monitor.onMisfire(TaskResultImpl.<OUT>builder()
+                                        .setExternalId(jobData.externalId())
+                                        .setTick(state.tick())
+                                        .setAvailableAt(state.availableAt())
+                                        .setTriggeredAt(triggerAt)
+                                        .build());
     }
 
     @SuppressWarnings("unchecked")
     protected final void onResult(@NotNull AsyncResult<Object> asyncResult) {
         state.markIdle();
         final Instant finishedAt = Instant.now();
         if (asyncResult.failed()) {
-            LOGGER.warn(genMsg(state.tick(), state.round(), finishedAt, "Internal execution error"),
-                        asyncResult.cause());
+            final Throwable cause = asyncResult.cause();
+            LOGGER.warn(genMsg(state.tick(), state.round(), finishedAt, "Internal execution error"), cause);
+            monitor.onEach(TaskResultImpl.<OUT>builder()
+                                         .setExternalId(jobData.externalId())
+                                         .setAvailableAt(state.availableAt())
+                                         .setTick(state.tick())
+                                         .setRound(state.round())
+                                         .setFinishedAt(finishedAt)
+                                         .setError(state.addError(state.round(), cause))
+                                         .build());
         }
         TaskExecutionContextInternal<OUT> executionContext = (TaskExecutionContextInternal<OUT>) asyncResult.result();
         if (asyncResult.succeeded()) {
-            trace(state.tick(), executionContext.round(), finishedAt, "Received the task result");
+            trace(finishedAt, "Received the task result");
             monitor.onEach(TaskResultImpl.<OUT>builder()
                                          .setExternalId(jobData.externalId())
                                          .setAvailableAt(state.availableAt())
@@ -242,14 +259,13 @@ protected final void onResult(@NotNull AsyncResult<Object> asyncResult) {
         }
         if (shouldStop(executionContext, state.round())) {
             doStop(state.timerId());
-            onCompleted();
         }
     }
 
     protected final void onCompleted() {
         state.markCompleted();
         final Instant completedAt = Instant.now();
-        trace(state.tick(), state.round(), completedAt, "The task execution is completed");
+        trace(completedAt, "The task execution is completed");
         monitor.onCompleted(TaskResultImpl.<OUT>builder()
                                           .setExternalId(jobData.externalId())
                                           .setAvailableAt(state.availableAt())
@@ -262,6 +278,12 @@ protected final void onCompleted() {
                                           .build());
     }
 
+    private TaskExecutionContextInternal<OUT> onExecute(@NotNull Promise<Object> promise,
+                                                        @NotNull TaskExecutionContextInternal<OUT> executionContext) {
+        state.markExecuting();
+        return executionContext.setup(promise, Instant.now());
+    }
+
     private String genMsg(long tick, long round, @NotNull Instant at, @NotNull String event) {
         return "TaskExecutor[" + tick + "][" + round + "][" + at + "]::[" + jobData.externalId() + "] - " + event;
     }

core/src/main/java/io/github/zero88/schedulerx/impl/AbstractTaskExecutorBuilder.java

@@ -3,6 +3,7 @@
 import java.util.Objects;
 import java.util.Optional;
 
+import org.jetbrains.annotations.ApiStatus.Internal;
 import org.jetbrains.annotations.NotNull;
 
 import io.github.zero88.schedulerx.JobData;
@@ -18,10 +19,13 @@
  * The base task executor builder
  */
 @SuppressWarnings("unchecked")
-public abstract class AbstractTaskExecutorBuilder<IN, OUT, T extends Trigger, E extends TriggerTaskExecutor<IN, OUT,
-                                                                                                               T>,
+@Internal
+// @formatter:off
+public abstract class AbstractTaskExecutorBuilder<IN, OUT, T extends Trigger,
+                                                     E extends TriggerTaskExecutor<IN, OUT, T>,
                                                      B extends TriggerTaskExecutorBuilder<IN, OUT, T, E, B>>
     implements TriggerTaskExecutorBuilder<IN, OUT, T, E, B> {
+// @formatter:on
 
     private Vertx vertx;
     private TaskExecutorMonitor<OUT> monitor;
@@ -46,9 +50,7 @@ public abstract class AbstractTaskExecutorBuilder<IN, OUT, T extends Trigger, E
     public @NotNull Task<IN, OUT> task() { return Objects.requireNonNull(task, "Task is required"); }
 
     @Override
-    public @NotNull JobData<IN> jobData() {
-        return Optional.ofNullable(jobData).orElseGet(JobData::empty);
-    }
+    public @NotNull JobData<IN> jobData() { return Optional.ofNullable(jobData).orElseGet(JobData::empty); }
 
     public @NotNull B setVertx(@NotNull Vertx vertx) {
         this.vertx = vertx;

core/src/main/java/io/github/zero88/schedulerx/impl/InternalTriggerContext.java

@@ -0,0 +1,27 @@
+package io.github.zero88.schedulerx.impl;
+
+import org.jetbrains.annotations.ApiStatus.Internal;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+import io.github.zero88.schedulerx.trigger.TriggerContext;
+
+@Internal
+public interface InternalTriggerContext extends TriggerContext {
+
+    boolean shouldRun();
+
+    static InternalTriggerContext create(boolean shouldRun, TriggerContext ctx) {
+        return new InternalTriggerContext() {
+            @Override
+            public boolean shouldRun() { return shouldRun; }
+
+            @Override
+            public @Nullable Object info() { return ctx.info(); }
+
+            @Override
+            public @NotNull String type() { return ctx.type(); }
+        };
+    }
+
+}