feat(api): task handler interceptors for cross-cutting concerns (#278)

api-impl/src/main/kotlin/dev/bpmcrafters/processengineapi/impl/task/AbstractTaskSubscriptionApiImpl.kt

@@ -2,6 +2,7 @@ package dev.bpmcrafters.processengineapi.impl.task
 
 import dev.bpmcrafters.processengineapi.Empty
 import dev.bpmcrafters.processengineapi.task.SubscribeForTaskCmd
+import dev.bpmcrafters.processengineapi.task.TaskHandlerInterceptor
 import dev.bpmcrafters.processengineapi.task.TaskSubscription
 import dev.bpmcrafters.processengineapi.task.TaskSubscriptionApi
 import dev.bpmcrafters.processengineapi.task.UnsubscribeFromTaskCmd
@@ -14,16 +15,22 @@ private val logger = KotlinLogging.logger {}
  * Abstract task subscription api implementation, using subscription repository and a list of completion strategies.
  * @since 0.0.2
  */
-abstract class AbstractTaskSubscriptionApiImpl(
-  private val subscriptionRepository: SubscriptionRepository
+abstract class AbstractTaskSubscriptionApiImpl @JvmOverloads constructor(
+  private val subscriptionRepository: SubscriptionRepository,
+  private val interceptors: List<TaskHandlerInterceptor> = emptyList()
 ) : TaskSubscriptionApi {
 
   override fun subscribeForTask(cmd: SubscribeForTaskCmd): CompletableFuture<TaskSubscription> {
     return TaskSubscriptionHandle(
       taskDescriptionKey = cmd.taskDescriptionKey,
       payloadDescription = cmd.payloadDescription,
       restrictions = cmd.restrictions,
-      action = cmd.action,
+      action = if (interceptors.isEmpty()) cmd.action else InterceptingTaskHandler(
+        delegate = cmd.action,
+        interceptors = interceptors,
+        taskDescriptionKey = cmd.taskDescriptionKey,
+        taskType = cmd.taskType
+      ),
       taskType = cmd.taskType,
       termination = cmd.termination
     ).let {

api-impl/src/main/kotlin/dev/bpmcrafters/processengineapi/impl/task/InterceptingTaskHandler.kt

@@ -0,0 +1,42 @@
+package dev.bpmcrafters.processengineapi.impl.task
+
+import dev.bpmcrafters.processengineapi.task.TaskHandler
+import dev.bpmcrafters.processengineapi.task.TaskHandlerInterceptor
+import dev.bpmcrafters.processengineapi.task.TaskHandlerInterceptorChain
+import dev.bpmcrafters.processengineapi.task.TaskHandlerInterceptorContext
+import dev.bpmcrafters.processengineapi.task.TaskHandlerOutcome
+import dev.bpmcrafters.processengineapi.task.TaskInformation
+import dev.bpmcrafters.processengineapi.task.TaskType
+
+/**
+ * Wraps a [TaskHandler] and runs a list of [TaskHandlerInterceptor]s around its execution.
+ * Interceptors run in the order of the given list; the wrapped [delegate] is invoked last.
+ * @since 1.7
+ */
+class InterceptingTaskHandler(
+  private val delegate: TaskHandler,
+  private val interceptors: List<TaskHandlerInterceptor>,
+  private val taskDescriptionKey: String?,
+  private val taskType: TaskType
+) : TaskHandler {
+
+  override fun accept(taskInformation: TaskInformation, payload: Map<String, Any?>) {
+    val context = TaskHandlerInterceptorContext(taskInformation, payload, taskDescriptionKey, taskType)
+    TaskHandlerOutcomeRecorder.recordingFor(taskInformation.taskId) {
+      proceed(context, 0, taskInformation, payload)
+    }
+  }
+
+  private fun proceed(
+    context: TaskHandlerInterceptorContext,
+    index: Int,
+    taskInformation: TaskInformation,
+    payload: Map<String, Any?>
+  ): TaskHandlerOutcome =
+    if (index < interceptors.size) {
+      interceptors[index].intercept(context) { proceed(context, index + 1, taskInformation, payload) }
+    } else {
+      delegate.accept(taskInformation, payload)
+      TaskHandlerOutcomeRecorder.recordedOutcome()
+    }
+}

api-impl/src/main/kotlin/dev/bpmcrafters/processengineapi/impl/task/OutcomeRecordingServiceTaskCompletionApi.kt

@@ -0,0 +1,37 @@
+package dev.bpmcrafters.processengineapi.impl.task
+
+import dev.bpmcrafters.processengineapi.Empty
+import dev.bpmcrafters.processengineapi.task.CompleteTaskByErrorCmd
+import dev.bpmcrafters.processengineapi.task.CompleteTaskCmd
+import dev.bpmcrafters.processengineapi.task.FailTaskCmd
+import dev.bpmcrafters.processengineapi.task.ServiceTaskCompletionApi
+import dev.bpmcrafters.processengineapi.task.TaskHandlerInterceptor
+import dev.bpmcrafters.processengineapi.task.TaskHandlerOutcome
+import java.util.concurrent.CompletableFuture
+
+/**
+ * Decorates a [ServiceTaskCompletionApi] and reports every completion or failure requested by a
+ * task handler as [TaskHandlerOutcome] to the [TaskHandlerInterceptor]s of the running execution.
+ *
+ * The outcome is only reported if the handler calls this API in the thread executing the handler.
+ * @since 1.7
+ */
+class OutcomeRecordingServiceTaskCompletionApi(
+  private val delegate: ServiceTaskCompletionApi
+) : ServiceTaskCompletionApi {
+
+  override fun completeTask(cmd: CompleteTaskCmd): CompletableFuture<Empty> =
+    delegate.completeTask(cmd).also {
+      TaskHandlerOutcomeRecorder.record(cmd.taskId, TaskHandlerOutcome.Completed)
+    }
+
+  override fun completeTaskByError(cmd: CompleteTaskByErrorCmd): CompletableFuture<Empty> =
+    delegate.completeTaskByError(cmd).also {
+      TaskHandlerOutcomeRecorder.record(cmd.taskId, TaskHandlerOutcome.CompletedByError(cmd.errorCode, cmd.errorMessage))
+    }
+
+  override fun failTask(cmd: FailTaskCmd): CompletableFuture<Empty> =
+    delegate.failTask(cmd).also {
+      TaskHandlerOutcomeRecorder.record(cmd.taskId, TaskHandlerOutcome.Failed(cmd.reason, cmd.errorDetails))
+    }
+}

api-impl/src/main/kotlin/dev/bpmcrafters/processengineapi/impl/task/TaskHandlerOutcomeRecorder.kt

@@ -0,0 +1,32 @@
+package dev.bpmcrafters.processengineapi.impl.task
+
+import dev.bpmcrafters.processengineapi.task.TaskHandlerOutcome
+
+/**
+ * Keeps the [TaskHandlerOutcome] reported for the task currently handled by the executing thread.
+ * @since 1.7
+ */
+internal object TaskHandlerOutcomeRecorder {
+
+  private class Recording(val taskId: String) {
+    var outcome: TaskHandlerOutcome = TaskHandlerOutcome.Undetermined
+  }
+
+  private val currentRecording = ThreadLocal<Recording?>()
+
+  fun <T> recordingFor(taskId: String, execution: () -> T): T {
+    val enclosingRecording = currentRecording.get()
+    currentRecording.set(Recording(taskId))
+    try {
+      return execution()
+    } finally {
+      if (enclosingRecording == null) currentRecording.remove() else currentRecording.set(enclosingRecording)
+    }
+  }
+
+  fun record(taskId: String, outcome: TaskHandlerOutcome) {
+    currentRecording.get()?.takeIf { it.taskId == taskId }?.outcome = outcome
+  }
+
+  fun recordedOutcome(): TaskHandlerOutcome = currentRecording.get()?.outcome ?: TaskHandlerOutcome.Undetermined
+}

api-impl/src/test/kotlin/dev/bpmcrafters/processengineapi/impl/task/AbstractTaskSubscriptionApiImplTest.kt

@@ -4,6 +4,9 @@ import dev.bpmcrafters.processengineapi.CommonRestrictions
 import dev.bpmcrafters.processengineapi.MetaInfo
 import dev.bpmcrafters.processengineapi.MetaInfoAware
 import dev.bpmcrafters.processengineapi.task.SubscribeForTaskCmd
+import dev.bpmcrafters.processengineapi.task.TaskHandler
+import dev.bpmcrafters.processengineapi.task.TaskHandlerInterceptor
+import dev.bpmcrafters.processengineapi.task.TaskInformation
 import dev.bpmcrafters.processengineapi.task.TaskType
 import dev.bpmcrafters.processengineapi.task.UnsubscribeFromTaskCmd
 import org.assertj.core.api.Assertions
@@ -13,7 +16,10 @@ internal class AbstractTaskSubscriptionApiImplTest {
 
   private val subscriptionRepository: SubscriptionRepository = InMemSubscriptionRepository()
   private val taskSubscriptionApi = MyTaskSubscriptionApiImpl(subscriptionRepository)
-  class MyTaskSubscriptionApiImpl(subscriptionRepository: SubscriptionRepository) : AbstractTaskSubscriptionApiImpl(subscriptionRepository) {
+  class MyTaskSubscriptionApiImpl(
+    subscriptionRepository: SubscriptionRepository,
+    interceptors: List<TaskHandlerInterceptor> = emptyList()
+  ) : AbstractTaskSubscriptionApiImpl(subscriptionRepository, interceptors) {
 
     override fun meta(instance: MetaInfoAware): MetaInfo {
       TODO("Not yet implemented")
@@ -46,4 +52,51 @@ internal class AbstractTaskSubscriptionApiImplTest {
 
     Assertions.assertThat(this.subscriptionRepository.getTaskSubscriptions()).isEmpty()
   }
+
+  @Test
+  fun `keeps original action when no interceptors are configured`() {
+    val action = TaskHandler { _, _ -> }
+
+    val handle = taskSubscriptionApi.subscribeForTask(
+        SubscribeForTaskCmd(
+            CommonRestrictions.builder().withProcessDefinitionKey("process-key").build(),
+            TaskType.EXTERNAL,
+            null,
+            setOf("var1"),
+            action,
+            { _ -> }
+        )
+    ).get() as TaskSubscriptionHandle
+
+    Assertions.assertThat(handle.action).isSameAs(action)
+  }
+
+  @Test
+  fun `wraps action so interceptor runs on delivery and unsubscribe still works`() {
+    val trace = mutableListOf<String>()
+    val interceptor = TaskHandlerInterceptor { _, chain ->
+      trace += "before"
+      chain.proceed().also { trace += "after" }
+    }
+    val api = MyTaskSubscriptionApiImpl(subscriptionRepository, listOf(interceptor))
+    val action = TaskHandler { _, _ -> trace += "action" }
+
+    val handle = api.subscribeForTask(
+        SubscribeForTaskCmd(
+            CommonRestrictions.builder().withProcessDefinitionKey("process-key").build(),
+            TaskType.EXTERNAL,
+            null,
+            setOf("var1"),
+            action,
+            { _ -> }
+        )
+    ).get() as TaskSubscriptionHandle
+
+    Assertions.assertThat(handle.action).isNotSameAs(action)
+    handle.action.accept(TaskInformation("task-id", emptyMap()), emptyMap())
+    Assertions.assertThat(trace).containsExactly("before", "action", "after")
+
+    api.unsubscribe(UnsubscribeFromTaskCmd(handle)).get()
+    Assertions.assertThat(subscriptionRepository.getTaskSubscriptions()).isEmpty()
+  }
 }

api-impl/src/test/kotlin/dev/bpmcrafters/processengineapi/impl/task/InterceptingTaskHandlerTest.kt

@@ -0,0 +1,77 @@
+package dev.bpmcrafters.processengineapi.impl.task
+
+import dev.bpmcrafters.processengineapi.task.TaskHandler
+import dev.bpmcrafters.processengineapi.task.TaskHandlerInterceptor
+import dev.bpmcrafters.processengineapi.task.TaskHandlerInterceptorContext
+import dev.bpmcrafters.processengineapi.task.TaskHandlerOutcome
+import dev.bpmcrafters.processengineapi.task.TaskInformation
+import dev.bpmcrafters.processengineapi.task.TaskType
+import org.assertj.core.api.Assertions.assertThat
+import org.assertj.core.api.Assertions.assertThatThrownBy
+import org.junit.jupiter.api.Test
+
+internal class InterceptingTaskHandlerTest {
+
+  private val taskInformation = TaskInformation("task-id", mapOf("meta-key" to "meta-value"))
+  private val payload = mapOf("var1" to "value1")
+
+  @Test
+  fun `runs interceptors around delegate in list order`() {
+    val trace = mutableListOf<String>()
+    val first = TaskHandlerInterceptor { _, chain ->
+      trace += "first-before"
+      chain.proceed().also { trace += "first-after" }
+    }
+    val second = TaskHandlerInterceptor { _, chain ->
+      trace += "second-before"
+      chain.proceed().also { trace += "second-after" }
+    }
+    val delegate = TaskHandler { _, _ -> trace += "delegate" }
+
+    InterceptingTaskHandler(delegate, listOf(first, second), "task-key", TaskType.EXTERNAL)
+      .accept(taskInformation, payload)
+
+    assertThat(trace).containsExactly(
+      "first-before", "second-before", "delegate", "second-after", "first-after"
+    )
+  }
+
+  @Test
+  fun `provides context to interceptors`() {
+    var captured: TaskHandlerInterceptorContext? = null
+    val interceptor = TaskHandlerInterceptor { context, chain ->
+      captured = context
+      chain.proceed()
+    }
+
+    InterceptingTaskHandler(TaskHandler { _, _ -> }, listOf(interceptor), "task-key", TaskType.USER)
+      .accept(taskInformation, payload)
+
+    assertThat(captured).isEqualTo(
+      TaskHandlerInterceptorContext(taskInformation, payload, "task-key", TaskType.USER)
+    )
+  }
+
+  @Test
+  fun `interceptor without proceed prevents delegate invocation`() {
+    var delegateCalled = false
+    val interceptor = TaskHandlerInterceptor { _, _ -> TaskHandlerOutcome.Undetermined }
+    val delegate = TaskHandler { _, _ -> delegateCalled = true }
+
+    InterceptingTaskHandler(delegate, listOf(interceptor), "task-key", TaskType.EXTERNAL)
+      .accept(taskInformation, payload)
+
+    assertThat(delegateCalled).isFalse()
+  }
+
+  @Test
+  fun `delegate exception propagates through the chain`() {
+    val interceptor = TaskHandlerInterceptor { _, chain -> chain.proceed() }
+    val delegate = TaskHandler { _, _ -> throw IllegalStateException("boom") }
+
+    assertThatThrownBy {
+      InterceptingTaskHandler(delegate, listOf(interceptor), "task-key", TaskType.EXTERNAL)
+        .accept(taskInformation, payload)
+    }.isInstanceOf(IllegalStateException::class.java).hasMessage("boom")
+  }
+}