cla
}
/**
- * Build a call step for functions that need {@code WorkflowContextData} as the first parameter.
+ * Build a call step for functions that need {@link WorkflowContextData} as the first parameter.
* The DSL wraps it as a {@link JavaContextFunction} and injects the runtime context.
*
* Signature expected: {@code (ctx, payload) -> result}
@@ -297,7 +300,7 @@ public static FuncCallStep function(Function fn, Class cla
* @param result type
* @return a call step
*/
- public static FuncCallStep withContext(CtxBiFunction fn, Class in) {
+ public static FuncCallStep withContext(JavaContextFunction fn, Class in) {
return withContext(null, fn, in);
}
@@ -319,7 +322,7 @@ public static FuncCallStep withInstanceId(
}
/**
- * Named variant of {@link #withContext(CtxBiFunction, Class)}.
+ * Named variant of {@link #withContext(JavaContextFunction, Class)}.
*
* @param name task name
* @param fn context-aware bi-function
@@ -329,9 +332,40 @@ public static FuncCallStep withInstanceId(
* @return a named call step
*/
public static FuncCallStep withContext(
- String name, CtxBiFunction fn, Class in) {
- JavaContextFunction jcf = (payload, wctx) -> fn.apply(wctx, payload);
- return new FuncCallStep<>(name, jcf, in);
+ String name, JavaContextFunction fn, Class in) {
+ return new FuncCallStep<>(name, fn, in);
+ }
+
+ /**
+ * Build a call step for functions that need {@link WorkflowContextData} and {@link
+ * io.serverlessworkflow.impl.TaskContextData} as the first and second parameter. The DSL wraps it
+ * as a {@link JavaFilterFunction} and injects the runtime context.
+ *
+ * Signature expected: {@code (wctx, tctx, payload) -> result}
+ *
+ * @param fn context-aware bi-function
+ * @param in payload input class
+ * @param input type
+ * @param result type
+ * @return a call step
+ */
+ public static FuncCallStep withFilter(JavaFilterFunction fn, Class in) {
+ return withFilter(null, fn, in);
+ }
+
+ /**
+ * Named variant of {@link #withFilter(JavaFilterFunction, Class)}.
+ *
+ * @param name task name
+ * @param fn context-aware bi-function
+ * @param in payload input class
+ * @param input type
+ * @param result type
+ * @return a named call step
+ */
+ public static FuncCallStep withFilter(
+ String name, JavaFilterFunction fn, Class in) {
+ return new FuncCallStep<>(name, fn, in);
}
/**
@@ -350,6 +384,38 @@ public static FuncCallStep withInstanceId(
return new FuncCallStep<>(name, jcf, in);
}
+ /**
+ * Builds a composition of the current workflow instance id and the definition of the task
+ * position as a JSON pointer.
+ */
+ static String defaultUniqueId(WorkflowContextData wctx, TaskContextData tctx) {
+ return String.format("%s-%s", wctx.instanceData().id(), tctx.position().jsonPointer());
+ }
+
+ /**
+ * Build a call step for functions that expect a composition with the workflow instance id and the
+ * task position as the first parameter. The instance ID is extracted from the runtime context,
+ * the task position from the definition.
+ *
+ * Signature expected: {@code (uniqueId, payload) -> result}
+ *
+ * @param fn unique-id-aware bi-function
+ * @param in payload input class
+ * @param input type
+ * @param result type
+ * @return a call step
+ */
+ public static FuncCallStep withUniqueId(
+ String name, UniqueIdBiFunction fn, Class in) {
+ JavaFilterFunction jff =
+ (payload, wctx, tctx) -> fn.apply(defaultUniqueId(wctx, tctx), payload);
+ return new FuncCallStep<>(name, jff, in);
+ }
+
+ public static FuncCallStep withUniqueId(UniqueIdBiFunction fn, Class in) {
+ return withUniqueId(null, fn, in);
+ }
+
/**
* Create a fire-and-forget side-effect step (unnamed). The consumer receives the typed input.
*
@@ -387,12 +453,12 @@ public static ConsumeStep consume(String name, Consumer consumer, Clas
* @param result type
* @return a call step
*/
- public static FuncCallStep agent(InstanceIdBiFunction fn, Class in) {
- return withInstanceId(fn, in);
+ public static FuncCallStep agent(UniqueIdBiFunction fn, Class in) {
+ return withUniqueId(fn, in);
}
/**
- * Named agent-style sugar. See {@link #agent(InstanceIdBiFunction, Class)}.
+ * Named agent-style sugar. See {@link #agent(UniqueIdBiFunction, Class)}.
*
* @param name task name
* @param fn (instanceId, payload) -> result
@@ -402,8 +468,8 @@ public static FuncCallStep agent(InstanceIdBiFunction fn, Cla
* @return a named call step
*/
public static FuncCallStep agent(
- String name, InstanceIdBiFunction fn, Class in) {
- return withInstanceId(name, fn, in);
+ String name, UniqueIdBiFunction fn, Class in) {
+ return withUniqueId(name, fn, in);
}
/**
@@ -677,7 +743,7 @@ public static FuncTaskConfigurer switchWhenOrElse(
* switchWhenOrElse(".approved == true", "sendEmail", FlowDirectiveEnum.END)
*
*
- * The JQ expression is evaluated against the task input at runtime.
+ * The JQ expression is evaluated against the task input at runtime.
*/
public static FuncTaskConfigurer switchWhenOrElse(
String jqExpression, String thenTask, FlowDirectiveEnum otherwise) {
@@ -698,7 +764,7 @@ public static FuncTaskConfigurer switchWhenOrElse(
* switchWhenOrElse(".score >= 80", "pass", "fail")
*
*
- * The JQ expression is evaluated against the task input at runtime.
+ *
The JQ expression is evaluated against the task input at runtime.
*/
public static FuncTaskConfigurer switchWhenOrElse(
String jqExpression, String thenTask, String otherwiseTask) {
diff --git a/experimental/fluent/func/src/main/java/io/serverlessworkflow/fluent/func/dsl/UniqueIdBiFunction.java b/experimental/fluent/func/src/main/java/io/serverlessworkflow/fluent/func/dsl/UniqueIdBiFunction.java
new file mode 100644
index 00000000..229ed716
--- /dev/null
+++ b/experimental/fluent/func/src/main/java/io/serverlessworkflow/fluent/func/dsl/UniqueIdBiFunction.java
@@ -0,0 +1,30 @@
+/*
+ * Copyright 2020-Present The Serverless Workflow Specification Authors
+ *
+ * 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 io.serverlessworkflow.fluent.func.dsl;
+
+import java.util.function.BiFunction;
+
+/**
+ * Functions that expect a unique ID injection in runtime, typically an idempotent generated unique
+ * id based on the workflow instance id and task name.
+ *
+ * @param The task payload input
+ * @param The task result output
+ */
+@FunctionalInterface
+public interface UniqueIdBiFunction extends BiFunction {
+ R apply(String uniqueId, T object);
+}
diff --git a/experimental/fluent/func/src/test/java/io/serverlessworkflow/fluent/func/FuncDSLUniqueIdTest.java b/experimental/fluent/func/src/test/java/io/serverlessworkflow/fluent/func/FuncDSLUniqueIdTest.java
new file mode 100644
index 00000000..fe7a9b07
--- /dev/null
+++ b/experimental/fluent/func/src/test/java/io/serverlessworkflow/fluent/func/FuncDSLUniqueIdTest.java
@@ -0,0 +1,149 @@
+/*
+ * Copyright 2020-Present The Serverless Workflow Specification Authors
+ *
+ * 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 io.serverlessworkflow.fluent.func;
+
+import static io.serverlessworkflow.fluent.func.dsl.FuncDSL.agent;
+import static io.serverlessworkflow.fluent.func.dsl.FuncDSL.withUniqueId;
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+import io.serverlessworkflow.api.types.Task;
+import io.serverlessworkflow.api.types.TaskItem;
+import io.serverlessworkflow.api.types.Workflow;
+import io.serverlessworkflow.api.types.func.CallJava;
+import io.serverlessworkflow.api.types.func.JavaFilterFunction;
+import io.serverlessworkflow.fluent.func.dsl.UniqueIdBiFunction;
+import io.serverlessworkflow.impl.TaskContextData;
+import io.serverlessworkflow.impl.WorkflowContextData;
+import io.serverlessworkflow.impl.WorkflowInstanceData;
+import io.serverlessworkflow.impl.WorkflowPosition;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicReference;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Verifies that withUniqueId/agent wrap the user's function so that, at runtime, the first argument
+ * is a "unique id" composed as instanceId + "-" + jsonPointer (e.g., inst-123-/do/0/task).
+ */
+class FuncDSLUniqueIdTest {
+
+ @SuppressWarnings("unchecked")
+ private static JavaFilterFunction extractJavaFilterFunction(CallJava callJava) {
+ if (callJava instanceof CallJava.CallJavaFilterFunction f) {
+ return (JavaFilterFunction) f.function();
+ }
+ fail("CallTask is not a CallJavaFilterFunction; DSL contract may have changed.");
+ return null; // unreachable
+ }
+
+ @Test
+ @DisplayName(
+ "withUniqueId(name, fn, in) composes uniqueId = instanceId-jsonPointer and passes it")
+ void withUniqueId_uses_json_pointer_for_unique_id() throws Exception {
+ AtomicReference receivedUniqueId = new AtomicReference<>();
+ AtomicReference receivedPayload = new AtomicReference<>();
+
+ UniqueIdBiFunction fn =
+ (uniqueId, payload) -> {
+ receivedUniqueId.set(uniqueId);
+ receivedPayload.set(payload);
+ return payload.toUpperCase();
+ };
+
+ Workflow wf =
+ FuncWorkflowBuilder.workflow("wf-unique-named")
+ .tasks(withUniqueId("notify", fn, String.class))
+ .build();
+
+ List items = wf.getDo();
+ assertEquals(1, items.size(), "one task expected");
+ Task t = items.get(0).getTask();
+ assertNotNull(t.getCallTask(), "CallTask expected");
+
+ CallJava cj = (CallJava) t.getCallTask().get();
+ var jff = extractJavaFilterFunction(cj);
+ assertNotNull(jff, "JavaFilterFunction must be present for withUniqueId");
+
+ // Mockito stubs for runtime contexts
+ WorkflowInstanceData inst = mock(WorkflowInstanceData.class);
+ when(inst.id()).thenReturn("inst-123");
+
+ WorkflowContextData wctx = mock(WorkflowContextData.class);
+ when(wctx.instanceData()).thenReturn(inst);
+
+ // Use JSON Pointer for the unique component instead of task name
+ final String pointer = "/do/0/task";
+ WorkflowPosition pos = mock(WorkflowPosition.class);
+ when(pos.jsonPointer()).thenReturn(pointer);
+
+ TaskContextData tctx = mock(TaskContextData.class);
+ when(tctx.position()).thenReturn(pos);
+
+ Object result = jff.apply("hello", wctx, tctx);
+
+ assertEquals(
+ "inst-123-" + pointer, receivedUniqueId.get(), "uniqueId must be instanceId-jsonPointer");
+ assertEquals(
+ "hello", receivedPayload.get(), "payload should be forwarded to the user function");
+ assertEquals("HELLO", result, "wrapped function result should be returned");
+ }
+
+ @Test
+ @DisplayName("agent(fn, in) composes uniqueId = instanceId-jsonPointer and passes it")
+ void agent_uses_json_pointer_for_unique_id() throws Exception {
+ AtomicReference receivedUniqueId = new AtomicReference<>();
+ AtomicReference receivedPayload = new AtomicReference<>();
+
+ UniqueIdBiFunction fn =
+ (uniqueId, payload) -> {
+ receivedUniqueId.set(uniqueId);
+ receivedPayload.set(payload);
+ return payload + 1;
+ };
+
+ Workflow wf = FuncWorkflowBuilder.workflow("wf-agent").tasks(agent(fn, Integer.class)).build();
+
+ List items = wf.getDo();
+ assertEquals(1, items.size(), "one task expected");
+ Task t = items.get(0).getTask();
+ assertNotNull(t.getCallTask(), "CallTask expected");
+
+ CallJava cj = (CallJava) t.getCallTask().get();
+ var jff = extractJavaFilterFunction(cj);
+ assertNotNull(jff, "JavaFilterFunction must be present for agent/withUniqueId");
+
+ WorkflowInstanceData inst = mock(WorkflowInstanceData.class);
+ when(inst.id()).thenReturn("wf-999");
+
+ WorkflowContextData wctx = mock(WorkflowContextData.class);
+ when(wctx.instanceData()).thenReturn(inst);
+
+ final String pointer = "/do/0/task";
+ WorkflowPosition pos = mock(WorkflowPosition.class);
+ when(pos.jsonPointer()).thenReturn(pointer);
+
+ TaskContextData tctx = mock(TaskContextData.class);
+ when(tctx.position()).thenReturn(pos);
+
+ Object result = jff.apply(41, wctx, tctx);
+
+ assertEquals(
+ "wf-999-" + pointer, receivedUniqueId.get(), "uniqueId must be instanceId-jsonPointer");
+ assertEquals(41, receivedPayload.get(), "payload should be forwarded to the user function");
+ assertEquals(42, result, "wrapped function result should be returned");
+ }
+}
diff --git a/experimental/types/src/main/java/io/serverlessworkflow/api/types/func/CallJava.java b/experimental/types/src/main/java/io/serverlessworkflow/api/types/func/CallJava.java
index 971e3a9c..a9699b6b 100644
--- a/experimental/types/src/main/java/io/serverlessworkflow/api/types/func/CallJava.java
+++ b/experimental/types/src/main/java/io/serverlessworkflow/api/types/func/CallJava.java
@@ -54,6 +54,10 @@ public static CallJava function(JavaContextFunction function, Class
return new CallJavaContextFunction<>(function, Optional.ofNullable(inputClass));
}
+ public static CallJava function(JavaFilterFunction function, Class inputClass) {
+ return new CallJavaFilterFunction<>(function, Optional.ofNullable(inputClass));
+ }
+
public static class CallJavaConsumer extends CallJava {
private static final long serialVersionUID = 1L;
private final Consumer consumer;
@@ -95,8 +99,8 @@ public Optional> inputClass() {
public static class CallJavaContextFunction extends CallJava {
private static final long serialVersionUID = 1L;
- private JavaContextFunction function;
- private Optional> inputClass;
+ private final JavaContextFunction function;
+ private final Optional> inputClass;
public CallJavaContextFunction(
JavaContextFunction function, Optional> inputClass) {
@@ -113,6 +117,26 @@ public Optional> inputClass() {
}
}
+ public static class CallJavaFilterFunction extends CallJava {
+ private static final long serialVersionUID = 1L;
+ private final JavaFilterFunction function;
+ private final Optional> inputClass;
+
+ public CallJavaFilterFunction(
+ JavaFilterFunction function, Optional> inputClass) {
+ this.function = function;
+ this.inputClass = inputClass;
+ }
+
+ public JavaFilterFunction function() {
+ return function;
+ }
+
+ public Optional> inputClass() {
+ return inputClass;
+ }
+ }
+
public static class CallJavaLoopFunction extends CallJava {
private static final long serialVersionUID = 1L;