From 1039d5e98f47d5ffef766228569e55f0f8e33e76 Mon Sep 17 00:00:00 2001 From: Marco Schaeck Date: Wed, 16 Sep 2026 15:51:44 +0200 Subject: [PATCH 1/4] feat(api): add task handler interceptors for cross-cutting concerns Introduce TaskHandlerInterceptor, TaskHandlerInterceptorContext and TaskHandlerInterceptorChain in the Spring-free core so cross-cutting concerns (logging, tracing, audit) can wrap every task delivery for both registration paths (@ProcessEngineWorker and manual subscribe). AbstractTaskSubscriptionApiImpl optionally wraps the subscription action in an InterceptingTaskHandler; an empty interceptor list keeps the current behaviour unchanged. Refs #278 --- .../task/AbstractTaskSubscriptionApiImpl.kt | 13 +++- .../impl/task/InterceptingTaskHandler.kt | 39 ++++++++++ .../AbstractTaskSubscriptionApiImplTest.kt | 56 ++++++++++++- .../impl/task/InterceptingTaskHandlerTest.kt | 78 +++++++++++++++++++ .../task/TaskHandlerInterceptor.kt | 22 ++++++ .../task/TaskHandlerInterceptorChain.kt | 12 +++ .../task/TaskHandlerInterceptorContext.kt | 24 ++++++ docs/task-subscription-api.md | 37 +++++++++ 8 files changed, 277 insertions(+), 4 deletions(-) create mode 100644 api-impl/src/main/kotlin/dev/bpmcrafters/processengineapi/impl/task/InterceptingTaskHandler.kt create mode 100644 api-impl/src/test/kotlin/dev/bpmcrafters/processengineapi/impl/task/InterceptingTaskHandlerTest.kt create mode 100644 api/src/main/kotlin/dev/bpmcrafters/processengineapi/task/TaskHandlerInterceptor.kt create mode 100644 api/src/main/kotlin/dev/bpmcrafters/processengineapi/task/TaskHandlerInterceptorChain.kt create mode 100644 api/src/main/kotlin/dev/bpmcrafters/processengineapi/task/TaskHandlerInterceptorContext.kt diff --git a/api-impl/src/main/kotlin/dev/bpmcrafters/processengineapi/impl/task/AbstractTaskSubscriptionApiImpl.kt b/api-impl/src/main/kotlin/dev/bpmcrafters/processengineapi/impl/task/AbstractTaskSubscriptionApiImpl.kt index f23bc6d5..94acd2a5 100644 --- a/api-impl/src/main/kotlin/dev/bpmcrafters/processengineapi/impl/task/AbstractTaskSubscriptionApiImpl.kt +++ b/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,8 +15,9 @@ 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 = emptyList() ) : TaskSubscriptionApi { override fun subscribeForTask(cmd: SubscribeForTaskCmd): CompletableFuture { @@ -23,7 +25,12 @@ abstract class AbstractTaskSubscriptionApiImpl( 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 { diff --git a/api-impl/src/main/kotlin/dev/bpmcrafters/processengineapi/impl/task/InterceptingTaskHandler.kt b/api-impl/src/main/kotlin/dev/bpmcrafters/processengineapi/impl/task/InterceptingTaskHandler.kt new file mode 100644 index 00000000..88443425 --- /dev/null +++ b/api-impl/src/main/kotlin/dev/bpmcrafters/processengineapi/impl/task/InterceptingTaskHandler.kt @@ -0,0 +1,39 @@ +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.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, + private val taskDescriptionKey: String?, + private val taskType: TaskType +) : TaskHandler { + + override fun accept(taskInformation: TaskInformation, payload: Map) { + val context = TaskHandlerInterceptorContext(taskInformation, payload, taskDescriptionKey, taskType) + proceed(context, 0, taskInformation, payload) + } + + private fun proceed( + context: TaskHandlerInterceptorContext, + index: Int, + taskInformation: TaskInformation, + payload: Map + ) { + if (index < interceptors.size) { + interceptors[index].intercept(context) { proceed(context, index + 1, taskInformation, payload) } + } else { + delegate.accept(taskInformation, payload) + } + } +} diff --git a/api-impl/src/test/kotlin/dev/bpmcrafters/processengineapi/impl/task/AbstractTaskSubscriptionApiImplTest.kt b/api-impl/src/test/kotlin/dev/bpmcrafters/processengineapi/impl/task/AbstractTaskSubscriptionApiImplTest.kt index 8a9f48e0..ffce3130 100644 --- a/api-impl/src/test/kotlin/dev/bpmcrafters/processengineapi/impl/task/AbstractTaskSubscriptionApiImplTest.kt +++ b/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 = emptyList() + ) : AbstractTaskSubscriptionApiImpl(subscriptionRepository, interceptors) { override fun meta(instance: MetaInfoAware): MetaInfo { TODO("Not yet implemented") @@ -46,4 +52,52 @@ 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() + val interceptor = TaskHandlerInterceptor { _, chain -> + trace += "before" + chain.proceed() + 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() + } } diff --git a/api-impl/src/test/kotlin/dev/bpmcrafters/processengineapi/impl/task/InterceptingTaskHandlerTest.kt b/api-impl/src/test/kotlin/dev/bpmcrafters/processengineapi/impl/task/InterceptingTaskHandlerTest.kt new file mode 100644 index 00000000..9576e135 --- /dev/null +++ b/api-impl/src/test/kotlin/dev/bpmcrafters/processengineapi/impl/task/InterceptingTaskHandlerTest.kt @@ -0,0 +1,78 @@ +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.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() + val first = TaskHandlerInterceptor { _, chain -> + trace += "first-before" + chain.proceed() + trace += "first-after" + } + val second = TaskHandlerInterceptor { _, chain -> + trace += "second-before" + chain.proceed() + 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 { _, _ -> } + 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") + } +} diff --git a/api/src/main/kotlin/dev/bpmcrafters/processengineapi/task/TaskHandlerInterceptor.kt b/api/src/main/kotlin/dev/bpmcrafters/processengineapi/task/TaskHandlerInterceptor.kt new file mode 100644 index 00000000..489c7a5e --- /dev/null +++ b/api/src/main/kotlin/dev/bpmcrafters/processengineapi/task/TaskHandlerInterceptor.kt @@ -0,0 +1,22 @@ +package dev.bpmcrafters.processengineapi.task + +/** + * Intercepts the execution of a [TaskHandler] of a subscription. + * + * Interceptors wrap the call `subscription.action.accept(taskInformation, payload)` performed by + * every delivery. They run in the delivery thread, inside the delivery's try/catch, and apply to + * every [TaskType]. An interceptor may filter on [TaskHandlerInterceptorContext.taskType]. + * + * @since 1.7 + */ +fun interface TaskHandlerInterceptor { + /** + * Intercepts the task handler execution. + * Call [TaskHandlerInterceptorChain.proceed] to continue with the next interceptor and finally + * the wrapped handler. Skipping the call prevents the handler from being invoked. An exception + * thrown here propagates like a handler exception. + * @param context context of the intercepted execution. + * @param chain chain to proceed with. + */ + fun intercept(context: TaskHandlerInterceptorContext, chain: TaskHandlerInterceptorChain) +} diff --git a/api/src/main/kotlin/dev/bpmcrafters/processengineapi/task/TaskHandlerInterceptorChain.kt b/api/src/main/kotlin/dev/bpmcrafters/processengineapi/task/TaskHandlerInterceptorChain.kt new file mode 100644 index 00000000..306dde1f --- /dev/null +++ b/api/src/main/kotlin/dev/bpmcrafters/processengineapi/task/TaskHandlerInterceptorChain.kt @@ -0,0 +1,12 @@ +package dev.bpmcrafters.processengineapi.task + +/** + * Chain to continue an intercepted [TaskHandler] execution. + * @since 1.7 + */ +fun interface TaskHandlerInterceptorChain { + /** + * Proceeds with the next interceptor and finally the wrapped handler. + */ + fun proceed() +} diff --git a/api/src/main/kotlin/dev/bpmcrafters/processengineapi/task/TaskHandlerInterceptorContext.kt b/api/src/main/kotlin/dev/bpmcrafters/processengineapi/task/TaskHandlerInterceptorContext.kt new file mode 100644 index 00000000..4cefa81c --- /dev/null +++ b/api/src/main/kotlin/dev/bpmcrafters/processengineapi/task/TaskHandlerInterceptorContext.kt @@ -0,0 +1,24 @@ +package dev.bpmcrafters.processengineapi.task + +/** + * Immutable context of an intercepted [TaskHandler] execution. + * @since 1.7 + */ +data class TaskHandlerInterceptorContext( + /** + * Information about the task being delivered. + */ + val taskInformation: TaskInformation, + /** + * Payload variables delivered to the handler. + */ + val payload: Map, + /** + * Task description key of the subscription, if any. + */ + val taskDescriptionKey: String?, + /** + * Type of the intercepted task. + */ + val taskType: TaskType +) diff --git a/docs/task-subscription-api.md b/docs/task-subscription-api.md index c333f402..c3cfd1b0 100644 --- a/docs/task-subscription-api.md +++ b/docs/task-subscription-api.md @@ -52,4 +52,41 @@ public class TaskSubscriber { } +``` + +## Intercepting task handlers + +Cross-cutting concerns such as logging, tracing, or auditing can wrap every task delivery via a +`TaskHandlerInterceptor`. An interceptor wraps the call of the subscription action for both +registration paths (`@ProcessEngineWorker` and manual `subscribeForTask`) and applies to every +`TaskType`. Filter on `context.getTaskType()` if an interceptor should only act on a specific type. + +Interceptors run in the delivery thread, inside the delivery's try/catch. An exception thrown by an +interceptor is treated like a handler exception and leads to failure handling or a retry by the +adapter. The context is immutable; interceptors observe the delivery but do not mutate payload or +task information. + +Register interceptors as Spring beans in the adapter starter. The order follows `@Order`/`Ordered`. + +```java + +@Component +@Slf4j +@Order(0) +public class LoggingTaskHandlerInterceptor implements TaskHandlerInterceptor { + + @Override + public void intercept(TaskHandlerInterceptorContext context, TaskHandlerInterceptorChain chain) { + log.info("[TASK]: Handling {} of type {}", context.getTaskInformation().getTaskId(), context.getTaskType()); + try { + chain.proceed(); + log.info("[TASK]: Handled {}", context.getTaskInformation().getTaskId()); + } catch (RuntimeException e) { + log.error("[TASK]: Failed handling {}", context.getTaskInformation().getTaskId(), e); + throw e; + } + } +} + + ``` From 84dd6aa9e7a2205025db38158a875a686994f0a4 Mon Sep 17 00:00:00 2001 From: Marco Schaeck Date: Wed, 16 Sep 2026 15:53:09 +0200 Subject: [PATCH 2/4] docs(api): clarify purpose of TaskHandlerInterceptor --- .../processengineapi/task/TaskHandlerInterceptor.kt | 3 +++ 1 file changed, 3 insertions(+) diff --git a/api/src/main/kotlin/dev/bpmcrafters/processengineapi/task/TaskHandlerInterceptor.kt b/api/src/main/kotlin/dev/bpmcrafters/processengineapi/task/TaskHandlerInterceptor.kt index 489c7a5e..e0249b21 100644 --- a/api/src/main/kotlin/dev/bpmcrafters/processengineapi/task/TaskHandlerInterceptor.kt +++ b/api/src/main/kotlin/dev/bpmcrafters/processengineapi/task/TaskHandlerInterceptor.kt @@ -3,6 +3,9 @@ package dev.bpmcrafters.processengineapi.task /** * Intercepts the execution of a [TaskHandler] of a subscription. * + * Lets you add cross-cutting concerns (logging, tracing, auditing) around every task execution + * without replacing the delivery layer or resorting to AOP. + * * Interceptors wrap the call `subscription.action.accept(taskInformation, payload)` performed by * every delivery. They run in the delivery thread, inside the delivery's try/catch, and apply to * every [TaskType]. An interceptor may filter on [TaskHandlerInterceptorContext.taskType]. From e5556b14ee6930c5de9e45e90cc01db1af8e2e5c Mon Sep 17 00:00:00 2001 From: Marco Schaeck Date: Wed, 16 Sep 2026 16:05:35 +0200 Subject: [PATCH 3/4] test(api): cover TaskHandlerInterceptorContext --- .../task/TaskHandlerInterceptorContextTest.kt | 20 +++++++++++++++++++ 1 file changed, 20 insertions(+) create mode 100644 api/src/test/kotlin/dev/bpmcrafters/processengineapi/task/TaskHandlerInterceptorContextTest.kt diff --git a/api/src/test/kotlin/dev/bpmcrafters/processengineapi/task/TaskHandlerInterceptorContextTest.kt b/api/src/test/kotlin/dev/bpmcrafters/processengineapi/task/TaskHandlerInterceptorContextTest.kt new file mode 100644 index 00000000..0d5f173e --- /dev/null +++ b/api/src/test/kotlin/dev/bpmcrafters/processengineapi/task/TaskHandlerInterceptorContextTest.kt @@ -0,0 +1,20 @@ +package dev.bpmcrafters.processengineapi.task + +import org.assertj.core.api.Assertions.assertThat +import org.junit.jupiter.api.Test + +internal class TaskHandlerInterceptorContextTest { + + @Test + fun `exposes the intercepted execution details`() { + val taskInformation = TaskInformation("task-id", mapOf("meta-key" to "meta-value")) + val payload = mapOf("var1" to "value1") + + val context = TaskHandlerInterceptorContext(taskInformation, payload, "task-key", TaskType.EXTERNAL) + + assertThat(context.taskInformation).isEqualTo(taskInformation) + assertThat(context.payload).isEqualTo(payload) + assertThat(context.taskDescriptionKey).isEqualTo("task-key") + assertThat(context.taskType).isEqualTo(TaskType.EXTERNAL) + } +} From c50d8dcd634e58fc1f868f96604d887a63131ca0 Mon Sep 17 00:00:00 2001 From: Marco Schaeck Date: Wed, 7 Oct 2026 13:40:08 +0200 Subject: [PATCH 4/4] feat(api): report task handler outcome to interceptors Interceptors receive a TaskHandlerOutcome from chain.proceed(), so a handler that fails the task or completes it by BPMN error itself is no longer seen as a normal return. Adds ACTIVITY_INSTANCE_ID as meta key. Refs #278 --- .../impl/task/InterceptingTaskHandler.kt | 9 +- ...utcomeRecordingServiceTaskCompletionApi.kt | 37 ++++++ .../impl/task/TaskHandlerOutcomeRecorder.kt | 32 +++++ .../AbstractTaskSubscriptionApiImplTest.kt | 3 +- .../impl/task/InterceptingTaskHandlerTest.kt | 9 +- ...meRecordingServiceTaskCompletionApiTest.kt | 116 ++++++++++++++++++ .../processengineapi/CommonRestrictions.kt | 5 + .../task/TaskHandlerInterceptor.kt | 7 +- .../task/TaskHandlerInterceptorChain.kt | 3 +- .../task/TaskHandlerOutcome.kt | 48 ++++++++ docs/task-subscription-api.md | 18 ++- 11 files changed, 272 insertions(+), 15 deletions(-) create mode 100644 api-impl/src/main/kotlin/dev/bpmcrafters/processengineapi/impl/task/OutcomeRecordingServiceTaskCompletionApi.kt create mode 100644 api-impl/src/main/kotlin/dev/bpmcrafters/processengineapi/impl/task/TaskHandlerOutcomeRecorder.kt create mode 100644 api-impl/src/test/kotlin/dev/bpmcrafters/processengineapi/impl/task/OutcomeRecordingServiceTaskCompletionApiTest.kt create mode 100644 api/src/main/kotlin/dev/bpmcrafters/processengineapi/task/TaskHandlerOutcome.kt diff --git a/api-impl/src/main/kotlin/dev/bpmcrafters/processengineapi/impl/task/InterceptingTaskHandler.kt b/api-impl/src/main/kotlin/dev/bpmcrafters/processengineapi/impl/task/InterceptingTaskHandler.kt index 88443425..c416992b 100644 --- a/api-impl/src/main/kotlin/dev/bpmcrafters/processengineapi/impl/task/InterceptingTaskHandler.kt +++ b/api-impl/src/main/kotlin/dev/bpmcrafters/processengineapi/impl/task/InterceptingTaskHandler.kt @@ -4,6 +4,7 @@ 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 @@ -21,7 +22,9 @@ class InterceptingTaskHandler( override fun accept(taskInformation: TaskInformation, payload: Map) { val context = TaskHandlerInterceptorContext(taskInformation, payload, taskDescriptionKey, taskType) - proceed(context, 0, taskInformation, payload) + TaskHandlerOutcomeRecorder.recordingFor(taskInformation.taskId) { + proceed(context, 0, taskInformation, payload) + } } private fun proceed( @@ -29,11 +32,11 @@ class InterceptingTaskHandler( index: Int, taskInformation: TaskInformation, payload: Map - ) { + ): TaskHandlerOutcome = if (index < interceptors.size) { interceptors[index].intercept(context) { proceed(context, index + 1, taskInformation, payload) } } else { delegate.accept(taskInformation, payload) + TaskHandlerOutcomeRecorder.recordedOutcome() } - } } diff --git a/api-impl/src/main/kotlin/dev/bpmcrafters/processengineapi/impl/task/OutcomeRecordingServiceTaskCompletionApi.kt b/api-impl/src/main/kotlin/dev/bpmcrafters/processengineapi/impl/task/OutcomeRecordingServiceTaskCompletionApi.kt new file mode 100644 index 00000000..1c40bd25 --- /dev/null +++ b/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 = + delegate.completeTask(cmd).also { + TaskHandlerOutcomeRecorder.record(cmd.taskId, TaskHandlerOutcome.Completed) + } + + override fun completeTaskByError(cmd: CompleteTaskByErrorCmd): CompletableFuture = + delegate.completeTaskByError(cmd).also { + TaskHandlerOutcomeRecorder.record(cmd.taskId, TaskHandlerOutcome.CompletedByError(cmd.errorCode, cmd.errorMessage)) + } + + override fun failTask(cmd: FailTaskCmd): CompletableFuture = + delegate.failTask(cmd).also { + TaskHandlerOutcomeRecorder.record(cmd.taskId, TaskHandlerOutcome.Failed(cmd.reason, cmd.errorDetails)) + } +} diff --git a/api-impl/src/main/kotlin/dev/bpmcrafters/processengineapi/impl/task/TaskHandlerOutcomeRecorder.kt b/api-impl/src/main/kotlin/dev/bpmcrafters/processengineapi/impl/task/TaskHandlerOutcomeRecorder.kt new file mode 100644 index 00000000..89ddd1c6 --- /dev/null +++ b/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() + + fun 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 +} diff --git a/api-impl/src/test/kotlin/dev/bpmcrafters/processengineapi/impl/task/AbstractTaskSubscriptionApiImplTest.kt b/api-impl/src/test/kotlin/dev/bpmcrafters/processengineapi/impl/task/AbstractTaskSubscriptionApiImplTest.kt index ffce3130..846cdb8b 100644 --- a/api-impl/src/test/kotlin/dev/bpmcrafters/processengineapi/impl/task/AbstractTaskSubscriptionApiImplTest.kt +++ b/api-impl/src/test/kotlin/dev/bpmcrafters/processengineapi/impl/task/AbstractTaskSubscriptionApiImplTest.kt @@ -76,8 +76,7 @@ internal class AbstractTaskSubscriptionApiImplTest { val trace = mutableListOf() val interceptor = TaskHandlerInterceptor { _, chain -> trace += "before" - chain.proceed() - trace += "after" + chain.proceed().also { trace += "after" } } val api = MyTaskSubscriptionApiImpl(subscriptionRepository, listOf(interceptor)) val action = TaskHandler { _, _ -> trace += "action" } diff --git a/api-impl/src/test/kotlin/dev/bpmcrafters/processengineapi/impl/task/InterceptingTaskHandlerTest.kt b/api-impl/src/test/kotlin/dev/bpmcrafters/processengineapi/impl/task/InterceptingTaskHandlerTest.kt index 9576e135..3820b79f 100644 --- a/api-impl/src/test/kotlin/dev/bpmcrafters/processengineapi/impl/task/InterceptingTaskHandlerTest.kt +++ b/api-impl/src/test/kotlin/dev/bpmcrafters/processengineapi/impl/task/InterceptingTaskHandlerTest.kt @@ -3,6 +3,7 @@ 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 @@ -19,13 +20,11 @@ internal class InterceptingTaskHandlerTest { val trace = mutableListOf() val first = TaskHandlerInterceptor { _, chain -> trace += "first-before" - chain.proceed() - trace += "first-after" + chain.proceed().also { trace += "first-after" } } val second = TaskHandlerInterceptor { _, chain -> trace += "second-before" - chain.proceed() - trace += "second-after" + chain.proceed().also { trace += "second-after" } } val delegate = TaskHandler { _, _ -> trace += "delegate" } @@ -56,7 +55,7 @@ internal class InterceptingTaskHandlerTest { @Test fun `interceptor without proceed prevents delegate invocation`() { var delegateCalled = false - val interceptor = TaskHandlerInterceptor { _, _ -> } + val interceptor = TaskHandlerInterceptor { _, _ -> TaskHandlerOutcome.Undetermined } val delegate = TaskHandler { _, _ -> delegateCalled = true } InterceptingTaskHandler(delegate, listOf(interceptor), "task-key", TaskType.EXTERNAL) diff --git a/api-impl/src/test/kotlin/dev/bpmcrafters/processengineapi/impl/task/OutcomeRecordingServiceTaskCompletionApiTest.kt b/api-impl/src/test/kotlin/dev/bpmcrafters/processengineapi/impl/task/OutcomeRecordingServiceTaskCompletionApiTest.kt new file mode 100644 index 00000000..eff582b1 --- /dev/null +++ b/api-impl/src/test/kotlin/dev/bpmcrafters/processengineapi/impl/task/OutcomeRecordingServiceTaskCompletionApiTest.kt @@ -0,0 +1,116 @@ +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.TaskHandler +import dev.bpmcrafters.processengineapi.task.TaskHandlerInterceptor +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 +import java.util.concurrent.CompletableFuture + +internal class OutcomeRecordingServiceTaskCompletionApiTest { + + private val taskInformation = TaskInformation("task-id", emptyMap()) + private val delegatedCommands = mutableListOf() + private val completionApi = OutcomeRecordingServiceTaskCompletionApi(object : ServiceTaskCompletionApi { + override fun completeTask(cmd: CompleteTaskCmd) = delegated(cmd) + override fun completeTaskByError(cmd: CompleteTaskByErrorCmd) = delegated(cmd) + override fun failTask(cmd: FailTaskCmd) = delegated(cmd) + }) + + private fun delegated(cmd: Any): CompletableFuture { + delegatedCommands += cmd + return CompletableFuture.completedFuture(Empty) + } + + private fun outcomeSeenByInterceptorOf(handler: TaskHandler): TaskHandlerOutcome { + var seenOutcome: TaskHandlerOutcome? = null + val interceptor = TaskHandlerInterceptor { _, chain -> chain.proceed().also { seenOutcome = it } } + InterceptingTaskHandler(handler, listOf(interceptor), "task-key", TaskType.EXTERNAL) + .accept(taskInformation, emptyMap()) + return seenOutcome!! + } + + @Test + fun `reports failed task`() { + val cmd = FailTaskCmd("task-id", "reason", "details") + + val outcome = outcomeSeenByInterceptorOf { _, _ -> completionApi.failTask(cmd).get() } + + assertThat(outcome).isEqualTo(TaskHandlerOutcome.Failed("reason", "details")) + assertThat(delegatedCommands).containsExactly(cmd) + } + + @Test + fun `reports task completed by error`() { + val cmd = CompleteTaskByErrorCmd("task-id", "error-code", "message") + + val outcome = outcomeSeenByInterceptorOf { _, _ -> completionApi.completeTaskByError(cmd).get() } + + assertThat(outcome).isEqualTo(TaskHandlerOutcome.CompletedByError("error-code", "message")) + assertThat(delegatedCommands).containsExactly(cmd) + } + + @Test + fun `reports completed task`() { + val cmd = CompleteTaskCmd("task-id") { emptyMap() } + + val outcome = outcomeSeenByInterceptorOf { _, _ -> completionApi.completeTask(cmd).get() } + + assertThat(outcome).isEqualTo(TaskHandlerOutcome.Completed) + assertThat(delegatedCommands).containsExactly(cmd) + } + + @Test + fun `reports undetermined outcome if handler does not finish the task`() { + val outcome = outcomeSeenByInterceptorOf { _, _ -> } + + assertThat(outcome).isEqualTo(TaskHandlerOutcome.Undetermined) + } + + @Test + fun `ignores outcome of another task`() { + val outcome = outcomeSeenByInterceptorOf { _, _ -> + completionApi.failTask(FailTaskCmd("other-task-id", "reason", null)).get() + } + + assertThat(outcome).isEqualTo(TaskHandlerOutcome.Undetermined) + } + + @Test + fun `ignores outcome reported by another thread`() { + val outcome = outcomeSeenByInterceptorOf { _, _ -> + CompletableFuture.runAsync { completionApi.failTask(FailTaskCmd("task-id", "reason", null)).get() }.get() + } + + assertThat(outcome).isEqualTo(TaskHandlerOutcome.Undetermined) + } + + @Test + fun `does not leak outcome of a throwing handler into the next execution`() { + assertThatThrownBy { + outcomeSeenByInterceptorOf { _, _ -> + completionApi.failTask(FailTaskCmd("task-id", "reason", null)).get() + throw IllegalStateException("boom") + } + }.isInstanceOf(IllegalStateException::class.java) + + assertThat(outcomeSeenByInterceptorOf { _, _ -> }).isEqualTo(TaskHandlerOutcome.Undetermined) + } + + @Test + fun `delegates without a running handler execution`() { + val cmd = FailTaskCmd("task-id", "reason", null) + + completionApi.failTask(cmd).get() + + assertThat(delegatedCommands).containsExactly(cmd) + } +} diff --git a/api/src/main/kotlin/dev/bpmcrafters/processengineapi/CommonRestrictions.kt b/api/src/main/kotlin/dev/bpmcrafters/processengineapi/CommonRestrictions.kt index 2043f35f..45fd75b9 100644 --- a/api/src/main/kotlin/dev/bpmcrafters/processengineapi/CommonRestrictions.kt +++ b/api/src/main/kotlin/dev/bpmcrafters/processengineapi/CommonRestrictions.kt @@ -11,6 +11,11 @@ object CommonRestrictions { */ const val ACTIVITY_ID = "activityId" + /** + * Id of the runtime instance of a BPMN activity. + */ + const val ACTIVITY_INSTANCE_ID = "activityInstanceId" + /** * Logical business key. */ diff --git a/api/src/main/kotlin/dev/bpmcrafters/processengineapi/task/TaskHandlerInterceptor.kt b/api/src/main/kotlin/dev/bpmcrafters/processengineapi/task/TaskHandlerInterceptor.kt index e0249b21..6d31cb29 100644 --- a/api/src/main/kotlin/dev/bpmcrafters/processengineapi/task/TaskHandlerInterceptor.kt +++ b/api/src/main/kotlin/dev/bpmcrafters/processengineapi/task/TaskHandlerInterceptor.kt @@ -18,8 +18,13 @@ fun interface TaskHandlerInterceptor { * Call [TaskHandlerInterceptorChain.proceed] to continue with the next interceptor and finally * the wrapped handler. Skipping the call prevents the handler from being invoked. An exception * thrown here propagates like a handler exception. + * + * A handler may report a failure or a BPMN error itself instead of throwing an exception. In this + * case [TaskHandlerInterceptorChain.proceed] returns normally and the returned [TaskHandlerOutcome] + * tells what happened. * @param context context of the intercepted execution. * @param chain chain to proceed with. + * @return outcome of the execution, usually the one returned by [TaskHandlerInterceptorChain.proceed]. */ - fun intercept(context: TaskHandlerInterceptorContext, chain: TaskHandlerInterceptorChain) + fun intercept(context: TaskHandlerInterceptorContext, chain: TaskHandlerInterceptorChain): TaskHandlerOutcome } diff --git a/api/src/main/kotlin/dev/bpmcrafters/processengineapi/task/TaskHandlerInterceptorChain.kt b/api/src/main/kotlin/dev/bpmcrafters/processengineapi/task/TaskHandlerInterceptorChain.kt index 306dde1f..d3a1fafc 100644 --- a/api/src/main/kotlin/dev/bpmcrafters/processengineapi/task/TaskHandlerInterceptorChain.kt +++ b/api/src/main/kotlin/dev/bpmcrafters/processengineapi/task/TaskHandlerInterceptorChain.kt @@ -7,6 +7,7 @@ package dev.bpmcrafters.processengineapi.task fun interface TaskHandlerInterceptorChain { /** * Proceeds with the next interceptor and finally the wrapped handler. + * @return outcome reported by the wrapped handler. */ - fun proceed() + fun proceed(): TaskHandlerOutcome } diff --git a/api/src/main/kotlin/dev/bpmcrafters/processengineapi/task/TaskHandlerOutcome.kt b/api/src/main/kotlin/dev/bpmcrafters/processengineapi/task/TaskHandlerOutcome.kt new file mode 100644 index 00000000..a1cd70d9 --- /dev/null +++ b/api/src/main/kotlin/dev/bpmcrafters/processengineapi/task/TaskHandlerOutcome.kt @@ -0,0 +1,48 @@ +package dev.bpmcrafters.processengineapi.task + +/** + * Outcome of an intercepted [TaskHandler] execution, as reported by the handler via the + * [ServiceTaskCompletionApi] in the thread executing the handler. + * @since 1.7 + */ +sealed interface TaskHandlerOutcome { + + /** + * The handler completed the task. + */ + data object Completed : TaskHandlerOutcome + + /** + * The handler completed the task by throwing a BPMN error. + */ + data class CompletedByError( + /** + * Error code. + */ + val errorCode: String, + /** + * Optional error details. + */ + val errorMessage: String? + ) : TaskHandlerOutcome + + /** + * The handler failed the task because of a technical failure. + */ + data class Failed( + /** + * Failure reason. + */ + val reason: String, + /** + * Optional failure details. + */ + val errorDetails: String? + ) : TaskHandlerOutcome + + /** + * The handler returned without reporting an outcome in the executing thread, for example because + * the task is completed later or by another thread, or because the handler was not invoked. + */ + data object Undetermined : TaskHandlerOutcome +} diff --git a/docs/task-subscription-api.md b/docs/task-subscription-api.md index c3cfd1b0..a0466a68 100644 --- a/docs/task-subscription-api.md +++ b/docs/task-subscription-api.md @@ -66,6 +66,13 @@ interceptor is treated like a handler exception and leads to failure handling or adapter. The context is immutable; interceptors observe the delivery but do not mutate payload or task information. +A handler may report a failure or a BPMN error itself instead of throwing an exception, as the +`process-engine-worker` does. `chain.proceed()` then returns normally, and the returned +`TaskHandlerOutcome` tells what happened: `Completed`, `CompletedByError`, `Failed`, or +`Undetermined`. The outcome is only known if the handler finishes the task through the +`ServiceTaskCompletionApi` in the delivery thread. A task finished later or by another thread +results in `Undetermined`. + Register interceptors as Spring beans in the adapter starter. The order follows `@Order`/`Ordered`. ```java @@ -76,11 +83,16 @@ Register interceptors as Spring beans in the adapter starter. The order follows public class LoggingTaskHandlerInterceptor implements TaskHandlerInterceptor { @Override - public void intercept(TaskHandlerInterceptorContext context, TaskHandlerInterceptorChain chain) { + public TaskHandlerOutcome intercept(TaskHandlerInterceptorContext context, TaskHandlerInterceptorChain chain) { log.info("[TASK]: Handling {} of type {}", context.getTaskInformation().getTaskId(), context.getTaskType()); try { - chain.proceed(); - log.info("[TASK]: Handled {}", context.getTaskInformation().getTaskId()); + TaskHandlerOutcome outcome = chain.proceed(); + if (outcome instanceof TaskHandlerOutcome.Failed failed) { + log.error("[TASK]: Failed handling {}: {}", context.getTaskInformation().getTaskId(), failed.getReason()); + } else { + log.info("[TASK]: Handled {}", context.getTaskInformation().getTaskId()); + } + return outcome; } catch (RuntimeException e) { log.error("[TASK]: Failed handling {}", context.getTaskInformation().getTaskId(), e); throw e;