Merge internal SchedulerClientEventListener into SchedulerListener

Author: kagkarlsson

db-scheduler/src/main/java/com/github/kagkarlsson/scheduler/ExecutePicked.java

@@ -37,7 +37,6 @@ class ExecutePicked implements Runnable {
   private static final Logger LOG = LoggerFactory.getLogger(ExecutePicked.class);
   private final Executor executor;
   private final TaskRepository taskRepository;
-  private SchedulerClientEventListener earlyExecutionListener;
   private final SchedulerClient schedulerClient;
   private final SchedulerListeners schedulerListeners;
   private final TaskResolver taskResolver;
@@ -50,7 +49,6 @@ class ExecutePicked implements Runnable {
   public ExecutePicked(
       Executor executor,
       TaskRepository taskRepository,
-      SchedulerClientEventListener earlyExecutionListener,
       SchedulerClient schedulerClient,
       SchedulerListeners schedulerListeners,
       TaskResolver taskResolver,
@@ -61,7 +59,6 @@ public ExecutePicked(
       Execution pickedExecution) {
     this.executor = executor;
     this.taskRepository = taskRepository;
-    this.earlyExecutionListener = earlyExecutionListener;
     this.schedulerClient = schedulerClient;
     this.schedulerListeners = schedulerListeners;
     this.taskResolver = taskResolver;
@@ -126,7 +123,7 @@ private void complete(
     try {
       completion.complete(
           completeEvent,
-          new ExecutionOperations(taskRepository, earlyExecutionListener, execution));
+          new ExecutionOperations(taskRepository, schedulerListeners, execution));
     } catch (Throwable e) {
       schedulerListeners.onSchedulerEvent(SchedulerEventType.COMPLETIONHANDLER_ERROR);
       schedulerListeners.onSchedulerEvent(SchedulerEventType.UNEXPECTED_ERROR);
@@ -156,7 +153,7 @@ private void failure(
       task.getFailureHandler()
           .onFailure(
               completeEvent,
-              new ExecutionOperations(taskRepository, earlyExecutionListener, execution));
+              new ExecutionOperations(taskRepository, schedulerListeners, execution));
     } catch (Throwable e) {
       schedulerListeners.onSchedulerEvent(SchedulerEventType.FAILUREHANDLER_ERROR);
       schedulerListeners.onSchedulerEvent(SchedulerEventType.UNEXPECTED_ERROR);

db-scheduler/src/main/java/com/github/kagkarlsson/scheduler/FetchCandidates.java

@@ -32,7 +32,6 @@ public class FetchCandidates implements PollStrategy {
   private final Executor executor;
   private final TaskRepository taskRepository;
   private final SchedulerClient schedulerClient;
-  private SchedulerClientEventListener earlyExecutionListener;
   private final SchedulerListeners schedulerListeners;
   private final SchedulerState schedulerState;
   private final ConfigurableLogger failureLogger;
@@ -49,7 +48,6 @@ public FetchCandidates(
       Executor executor,
       TaskRepository taskRepository,
       SchedulerClient schedulerClient,
-      SchedulerClientEventListener earlyExecutionListener,
       int threadpoolSize,
       SchedulerListeners schedulerListeners,
       SchedulerState schedulerState,
@@ -62,7 +60,6 @@ public FetchCandidates(
     this.executor = executor;
     this.taskRepository = taskRepository;
     this.schedulerClient = schedulerClient;
-    this.earlyExecutionListener = earlyExecutionListener;
     this.schedulerListeners = schedulerListeners;
     this.schedulerState = schedulerState;
     this.failureLogger = failureLogger;
@@ -105,7 +102,6 @@ public void run() {
                     new ExecutePicked(
                             executor,
                             taskRepository,
-                            earlyExecutionListener,
                             schedulerClient,
                             schedulerListeners,
                             taskResolver,

db-scheduler/src/main/java/com/github/kagkarlsson/scheduler/LockAndFetchCandidates.java

@@ -29,7 +29,6 @@ public class LockAndFetchCandidates implements PollStrategy {
   private final Executor executor;
   private final TaskRepository taskRepository;
   private final SchedulerClient schedulerClient;
-  private SchedulerClientEventListener earlyExecutionListener;
   private final SchedulerListeners schedulerListeners;
   private final TaskResolver taskResolver;
   private final SchedulerState schedulerState;
@@ -46,7 +45,6 @@ public LockAndFetchCandidates(
       Executor executor,
       TaskRepository taskRepository,
       SchedulerClient schedulerClient,
-      SchedulerClientEventListener earlyExecutionListener,
       int threadpoolSize,
       SchedulerListeners schedulerListeners,
       SchedulerState schedulerState,
@@ -59,7 +57,6 @@ public LockAndFetchCandidates(
     this.executor = executor;
     this.taskRepository = taskRepository;
     this.schedulerClient = schedulerClient;
-    this.earlyExecutionListener = earlyExecutionListener;
     this.schedulerListeners = schedulerListeners;
     this.taskResolver = taskResolver;
     this.schedulerState = schedulerState;
@@ -104,7 +101,6 @@ public void run() {
           new ExecutePicked(
               executor,
               taskRepository,
-              earlyExecutionListener,
               schedulerClient,
               schedulerListeners,
               taskResolver,

db-scheduler/src/main/java/com/github/kagkarlsson/scheduler/Scheduler.java

@@ -68,7 +68,6 @@ public class Scheduler implements SchedulerClient {
   private final Waiter heartbeatWaiter;
   final SettableSchedulerState schedulerState = new SettableSchedulerState();
   final ConfigurableLogger failureLogger;
-  private SchedulerClientEventListener earlyExecutionListener;
 
   protected Scheduler(
       Clock clock,
@@ -110,11 +109,10 @@ protected Scheduler(
     this.schedulerListeners = new SchedulerListeners(schedulerListeners);
     this.dueExecutor = dueExecutor;
     this.housekeeperExecutor = housekeeperExecutor;
-    earlyExecutionListener =
-        (enableImmediateExecution
-            ? new TriggerCheckForDueExecutions(schedulerState, clock, executeDueWaiter)
-            : SchedulerClientEventListener.NOOP);
-    delegate = new StandardSchedulerClient(clientTaskRepository, earlyExecutionListener, clock);
+    if (enableImmediateExecution) {
+      this.schedulerListeners.add(new TriggerCheckForDueExecutions(schedulerState, clock, executeDueWaiter));
+    }
+    delegate = new StandardSchedulerClient(clientTaskRepository, this.schedulerListeners, clock);
     this.failureLogger = ConfigurableLogger.create(LOG, logLevel, logStackTrace);
 
     if (pollingStrategyConfig.type == PollingStrategyConfig.Type.LOCK_AND_FETCH) {
@@ -124,7 +122,6 @@ protected Scheduler(
               executor,
               schedulerTaskRepository,
               this,
-              earlyExecutionListener,
               threadpoolSize,
               this.schedulerListeners,
               schedulerState,
@@ -140,7 +137,6 @@ protected Scheduler(
               executor,
               schedulerTaskRepository,
               this,
-              earlyExecutionListener,
               threadpoolSize,
               this.schedulerListeners,
               schedulerState,
@@ -368,7 +364,7 @@ protected void detectDeadExecutions() {
                     .deadExecution(
                         ExecutionComplete.failure(execution, now, now, null),
                         new ExecutionOperations(
-                            schedulerTaskRepository, earlyExecutionListener, execution));
+                            schedulerTaskRepository, schedulerListeners, execution));
               } else {
                 LOG.error(
                     "Failed to find implementation for task with name '{}' for detected dead execution. Either delete the execution from the databaser, or add an implementation for it.",

db-scheduler/src/main/java/com/github/kagkarlsson/scheduler/SchedulerClient.java

@@ -15,6 +15,7 @@
 
 import static java.util.Optional.ofNullable;
 
+import com.github.kagkarlsson.scheduler.event.SchedulerListeners;
 import com.github.kagkarlsson.scheduler.exceptions.TaskInstanceCurrentlyExecutingException;
 import com.github.kagkarlsson.scheduler.exceptions.TaskInstanceNotFoundException;
 import com.github.kagkarlsson.scheduler.jdbc.AutodetectJdbcCustomization;
@@ -239,18 +240,18 @@ class StandardSchedulerClient implements SchedulerClient {
     private static final Logger LOG = LoggerFactory.getLogger(StandardSchedulerClient.class);
     protected final TaskRepository taskRepository;
     private final Clock clock;
-    private SchedulerClientEventListener schedulerClientEventListener;
+    private final SchedulerListeners schedulerListeners;
 
     StandardSchedulerClient(TaskRepository taskRepository, Clock clock) {
-      this(taskRepository, SchedulerClientEventListener.NOOP, clock);
+      this(taskRepository, SchedulerListeners.NOOP, clock);
     }
 
     StandardSchedulerClient(
         TaskRepository taskRepository,
-        SchedulerClientEventListener schedulerClientEventListener,
+        SchedulerListeners schedulerListeners,
         Clock clock) {
       this.taskRepository = taskRepository;
-      this.schedulerClientEventListener = schedulerClientEventListener;
+      this.schedulerListeners = schedulerListeners;
       this.clock = clock;
     }
 
@@ -259,7 +260,7 @@ public <T> void schedule(TaskInstance<T> taskInstance, Instant executionTime) {
       boolean success =
           taskRepository.createIfNotExists(SchedulableInstance.of(taskInstance, executionTime));
       if (success) {
-        notifyListeners(ClientEvent.EventType.SCHEDULE, taskInstance, executionTime);
+        schedulerListeners.onExecutionScheduled(taskInstance, executionTime);
       }
     }
 
@@ -302,7 +303,7 @@ public <T> void reschedule(TaskInstanceId taskInstanceId, Instant newExecutionTi
         }
 
         if (success) {
-          notifyListeners(ClientEvent.EventType.RESCHEDULE, taskInstanceId, newExecutionTime);
+          schedulerListeners.onExecutionScheduled(taskInstanceId, newExecutionTime);
         }
       } else {
         throw new TaskInstanceNotFoundException(taskName, instanceId);
@@ -320,8 +321,6 @@ public void cancel(TaskInstanceId taskInstanceId) {
         }
 
         taskRepository.remove(execution.get());
-        notifyListeners(
-            ClientEvent.EventType.CANCEL, taskInstanceId, execution.get().executionTime);
       } else {
         throw new TaskInstanceNotFoundException(taskName, instanceId);
       }
@@ -366,16 +365,6 @@ public Optional<ScheduledExecution<Object>> getScheduledExecution(
       return e.map(oe -> new ScheduledExecution<>(Object.class, oe));
     }
 
-    private void notifyListeners(
-        ClientEvent.EventType eventType, TaskInstanceId taskInstanceId, Instant executionTime) {
-      try {
-        schedulerClientEventListener.newEvent(
-            new ClientEvent(
-                new ClientEvent.ClientEventContext(eventType, taskInstanceId, executionTime)));
-      } catch (Exception e) {
-        LOG.error("Error when notifying SchedulerClientEventListener.", e);
-      }
-    }
   }
 
   class SchedulerClientName implements SchedulerName {

db-scheduler/src/main/java/com/github/kagkarlsson/scheduler/SchedulerClientEventListener.java

@@ -13,11 +13,13 @@
  */
 package com.github.kagkarlsson.scheduler;
 
+import com.github.kagkarlsson.scheduler.event.AbstractSchedulerListener;
+import com.github.kagkarlsson.scheduler.task.TaskInstanceId;
 import java.time.Instant;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-class TriggerCheckForDueExecutions implements SchedulerClientEventListener {
+class TriggerCheckForDueExecutions extends AbstractSchedulerListener {
   private static final Logger LOG = LoggerFactory.getLogger(TriggerCheckForDueExecutions.class);
   private SchedulerState schedulerState;
   private Clock clock;
@@ -31,29 +33,22 @@ public TriggerCheckForDueExecutions(
   }
 
   @Override
-  public void newEvent(ClientEvent event) {
-    ClientEvent.ClientEventContext ctx = event.getContext();
-    ClientEvent.EventType eventType = ctx.getEventType();
-
+  public void onExecutionScheduled(
+      TaskInstanceId taskInstanceId, Instant scheduledToExecutionTime) {
     if (!schedulerState.isStarted() || schedulerState.isShuttingDown()) {
       LOG.debug(
           "Will not act on scheduling event for execution (task: '{}', id: '{}') as scheduler is starting or shutting down.",
-          ctx.getTaskInstanceId().getTaskName(),
-          ctx.getTaskInstanceId().getId());
+          taskInstanceId.getTaskName(),
+          taskInstanceId.getId());
       return;
     }
 
-    if (eventType == ClientEvent.EventType.SCHEDULE
-        || eventType == ClientEvent.EventType.RESCHEDULE) {
-
-      Instant scheduledToExecutionTime = ctx.getExecutionTime();
-      if (scheduledToExecutionTime.toEpochMilli() <= clock.now().toEpochMilli()) {
-        LOG.debug(
-            "Task-instance scheduled to run directly, triggering check for due executions (unless it is already running). Task: {}, instance: {}",
-            ctx.getTaskInstanceId().getTaskName(),
-            ctx.getTaskInstanceId().getId());
-        executeDueWaiter.wakeOrSkipNextWait();
-      }
+    if (scheduledToExecutionTime.toEpochMilli() <= clock.now().toEpochMilli()) {
+      LOG.debug(
+          "Task-instance scheduled to run directly, triggering check for due executions (unless it is already running). Task: {}, instance: {}",
+          taskInstanceId.getTaskName(),
+          taskInstanceId.getId());
+      executeDueWaiter.wakeOrSkipNextWait();
     }
   }
 }

db-scheduler/src/main/java/com/github/kagkarlsson/scheduler/TriggerCheckForDueExecutions.java

@@ -13,25 +13,22 @@
  */
 package com.github.kagkarlsson.scheduler.task;
 
-import com.github.kagkarlsson.scheduler.ClientEvent;
-import com.github.kagkarlsson.scheduler.ClientEvent.ClientEventContext;
-import com.github.kagkarlsson.scheduler.ClientEvent.EventType;
-import com.github.kagkarlsson.scheduler.SchedulerClientEventListener;
 import com.github.kagkarlsson.scheduler.TaskRepository;
+import com.github.kagkarlsson.scheduler.event.SchedulerListeners;
 import java.time.Instant;
 
 public class ExecutionOperations<T> {
 
   private final TaskRepository taskRepository;
-  private final SchedulerClientEventListener earlyExecutionListener;
+  private final SchedulerListeners schedulerListeners;
   private final Execution execution;
 
   public ExecutionOperations(
       TaskRepository taskRepository,
-      SchedulerClientEventListener earlyExecutionListener,
+      SchedulerListeners schedulerListeners,
       Execution execution) {
     this.taskRepository = taskRepository;
-    this.earlyExecutionListener = earlyExecutionListener;
+    this.schedulerListeners = schedulerListeners;
     this.execution = execution;
   }
 
@@ -81,8 +78,6 @@ public void reschedule(ExecutionComplete completed, Instant nextExecutionTime, T
 
   private void hintExecutionScheduled(TaskInstanceId taskInstanceId, Instant nextExecutionTime) {
     // Hint that a new execution was scheduled in-case we want to go check for it immediately
-    earlyExecutionListener.newEvent(
-        new ClientEvent(
-            new ClientEventContext(EventType.RESCHEDULE, taskInstanceId, nextExecutionTime)));
+    schedulerListeners.onExecutionScheduled(taskInstanceId, nextExecutionTime);
   }
 }