DD-2237 Made the PollingTaskExecutor more generic. Now, a task can be represe… (#33)

Changes to support DD-2237. Introduced a TaskScheduler interface that lets clients configure how tasks picked up by PollingTaskExecutor and created by TaskFactory are then scheduled (e.g., immediately executed or submitted to an executorservice).

Made sure the nextInput is called in its own db-transaction, so that a status update to the underlying record is not lost if the task execution fails.

Author: janvanmansum

pom.xml

@@ -25,7 +25,7 @@
         <relativePath />
     </parent>
     <artifactId>dans-java-utils</artifactId>
-    <version>2.11.2-SNAPSHOT</version>
+    <version>3.0.0-SNAPSHOT</version>
     <name>DANS Java Utility Classes</name>
     <inceptionYear>2021</inceptionYear>
     <scm>

src/main/java/nl/knaw/dans/lib/util/pollingtaskexec/ExecutorServiceTaskScheduler.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright (C) 2021 DANS - Data Archiving and Networked Services (info@dans.knaw.nl)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package nl.knaw.dans.lib.util.pollingtaskexec;
+
+import lombok.RequiredArgsConstructor;
+
+import java.util.concurrent.ExecutorService;
+
+/**
+ * A task scheduler implementation that schedules tasks for execution using an {@link ExecutorService}. This class allows for tasks to be executed asynchronously using the provided executor service.
+ *
+ */
+@RequiredArgsConstructor
+public class ExecutorServiceTaskScheduler implements TaskScheduler {
+    private final ExecutorService executorService;
+
+    @Override
+    public void schedule(Runnable task) {
+        executorService.execute(task);
+    }
+}

src/main/java/nl/knaw/dans/lib/util/pollingtaskexec/ImmediateTaskScheduler.java

@@ -0,0 +1,23 @@
+/*
+ * Copyright (C) 2021 DANS - Data Archiving and Networked Services (info@dans.knaw.nl)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package nl.knaw.dans.lib.util.pollingtaskexec;
+
+public class ImmediateTaskScheduler implements TaskScheduler {
+    @Override
+    public void schedule(Runnable task) {
+        task.run();
+    }
+}

src/main/java/nl/knaw/dans/lib/util/pollingtaskexec/PollingTaskExecutor.java

@@ -36,16 +36,21 @@
 @RequiredArgsConstructor
 public class PollingTaskExecutor<R> implements Managed {
     private final String name;
-    private final ScheduledExecutorService scheduler;
+    private final ScheduledExecutorService scheduledExecutorService;
     private final Duration pollingInterval;
     private final TaskSource<R> taskSource;
     private final TaskFactory<R> taskFactory;
+    private final TaskScheduler taskScheduler;
 
     private ScheduledFuture<?> future;
 
+    public PollingTaskExecutor(String name, ScheduledExecutorService scheduledExecutorService, Duration pollingInterval, TaskSource<R> taskSource, TaskFactory<R> taskFactory) {
+        this(name, scheduledExecutorService, pollingInterval, taskSource, taskFactory, new ImmediateTaskScheduler());
+    }
+
     /**
      * Copy constructor. The source executor must not be running. The purpose of this constructor is only to be able to wrap a PollingTaskExecutor in a UnitOfWorkAwareProxy. In general, no copies
-     * should be created of a PollingTaskExecutor, and in particular should the schedular not be shared among PollingTaskExecutors.
+     * should be created of a PollingTaskExecutor, and in particular should the scheduler not be shared among PollingTaskExecutors.
      *
      * @param other the source executor
      */
@@ -55,16 +60,17 @@ public PollingTaskExecutor(PollingTaskExecutor<R> other) {
             throw new IllegalArgumentException("Cannot copy a running executor");
         }
         this.name = other.name;
-        this.scheduler = other.scheduler;
+        this.scheduledExecutorService = other.scheduledExecutorService;
         this.pollingInterval = other.pollingInterval;
         this.taskSource = other.taskSource;
         this.taskFactory = other.taskFactory;
+        this.taskScheduler = other.taskScheduler;
     }
 
     @Override
     public void start() {
         long delayMs = Math.max(1L, pollingInterval.toMillis());
-        future = scheduler.scheduleWithFixedDelay(this::tick, 0, delayMs, TimeUnit.MILLISECONDS);
+        future = scheduledExecutorService.scheduleWithFixedDelay(this::tick, 0, delayMs, TimeUnit.MILLISECONDS);
         log.info("{} started; polling every {}", name, pollingInterval);
     }
 
@@ -73,24 +79,28 @@ public void stop() {
         if (future != null) {
             future.cancel(false);
         }
-        scheduler.shutdown();
+        scheduledExecutorService.shutdown();
         log.info("{} stopped", name);
     }
 
-    @UnitOfWork
     public void tick() {
         try {
-            Optional<R> next = taskSource.nextTask();
-            if (next.isEmpty()) {
+            Optional<R> input = getNextInput();
+            if (input.isEmpty()) {
                 return;
             }
-            R record = next.get();
-            log.debug("{}: found next task record: {}", name, record);
-            Runnable task = taskFactory.create(record);
-            task.run();
+            log.debug("{}: found next task input: {}", name, input.get());
+            Runnable task = taskFactory.create(input.get());
+            taskScheduler.schedule(task);
         }
         catch (Exception e) {
             log.error("{}: error while polling or running task", name, e);
         }
     }
+
+    // Must be protected for UnitOfWork to function.
+    @UnitOfWork
+    protected Optional<R> getNextInput() {
+        return taskSource.nextInput();
+    }
 }

src/main/java/nl/knaw/dans/lib/util/pollingtaskexec/TaskFactory.java

@@ -16,7 +16,7 @@
 package nl.knaw.dans.lib.util.pollingtaskexec;
 
 /**
- * Factory interface for creating {@link Runnable} tasks from a given record.
+ * Factory interface for creating {@link Runnable} tasks from a record.
  * This interface allows the decoupling of task creation logic from task execution logic.
  *
  * @param <R> the type of the record used to create a {@link Runnable} task

src/main/java/nl/knaw/dans/lib/util/pollingtaskexec/TaskScheduler.java

@@ -0,0 +1,26 @@
+/*
+ * Copyright (C) 2021 DANS - Data Archiving and Networked Services (info@dans.knaw.nl)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package nl.knaw.dans.lib.util.pollingtaskexec;
+
+public interface TaskScheduler {
+
+    /**
+     * Schedules the task for execution.
+     *
+     * @param task the task to schedule
+     */
+    void schedule(Runnable task);
+}

src/main/java/nl/knaw/dans/lib/util/pollingtaskexec/TaskSource.java

@@ -18,11 +18,12 @@
 import java.util.Optional;
 
 /**
- * Represents a source of tasks from which tasks can be fetched one at a time. Implementations of this interface are responsible for providing the logic to retrieve the next available task or
- * returning an empty result if no tasks are available.
+ * Represents a source of tasks from which tasks can be fetched. Implementations of this interface are responsible for providing the logic to retrieve the next available input that represents one
+ * task to be executed, returning an empty Optional if no tasks are available. When invoked by {@code PollingTaskExecutor}, this method is run within a {@code @UnitOfWork} so that any updates it
+ * makes to the database are visible to the resulting tasks.
  *
- * @param <R> the type of the tasks managed by this source
+ * @param <R> the type of the input used to create or schedule a task
  */
 public interface TaskSource<R> {
-    Optional<R> nextTask();
+    Optional<R> nextInput();
 }

src/test/java/nl/knaw/dans/lib/util/pollingtaskexec/PollingTaskExecutorTest.java

@@ -82,34 +82,34 @@ void tick_should_poll_source_create_task_and_run_it() {
         String record = "test-record";
         Runnable task = mock(Runnable.class);
 
-        when(taskSource.nextTask()).thenReturn(Optional.of(record));
+        when(taskSource.nextInput()).thenReturn(Optional.of(record));
         when(taskFactory.create(record)).thenReturn(task);
 
         executor.tick();
 
-        verify(taskSource).nextTask();
+        verify(taskSource).nextInput();
         verify(taskFactory).create(record);
         verify(task).run();
     }
 
     @Test
     void tick_should_do_nothing_if_source_is_empty() {
-        when(taskSource.nextTask()).thenReturn(Optional.empty());
+        when(taskSource.nextInput()).thenReturn(Optional.empty());
 
         executor.tick();
 
-        verify(taskSource).nextTask();
+        verify(taskSource).nextInput();
         verify(taskFactory, never()).create(any());
     }
 
     @Test
     void tick_should_catch_and_log_exceptions() {
-        when(taskSource.nextTask()).thenThrow(new RuntimeException("test exception"));
+        when(taskSource.nextInput()).thenThrow(new RuntimeException("test exception"));
 
         // Should not throw exception
         executor.tick();
 
-        verify(taskSource).nextTask();
+        verify(taskSource).nextInput();
     }
 
     @Test
@@ -119,12 +119,12 @@ void copy_constructor_should_copy_fields() {
         // Verify it works as expected by running a tick on the copy
         String record = "test-record";
         Runnable task = mock(Runnable.class);
-        when(taskSource.nextTask()).thenReturn(Optional.of(record));
+        when(taskSource.nextInput()).thenReturn(Optional.of(record));
         when(taskFactory.create(record)).thenReturn(task);
 
         copy.tick();
 
-        verify(taskSource).nextTask();
+        verify(taskSource).nextInput();
         verify(taskFactory).create(record);
         verify(task).run();
     }