Review FuncDSL

Signed-off-by: Ricardo Zanini <[email protected]>

Author: ricardozanini

experimental/fluent/func/pom.xml

@@ -21,6 +21,10 @@
             <groupId>io.serverlessworkflow</groupId>
             <artifactId>serverlessworkflow-experimental-types</artifactId>
         </dependency>
+        <dependency>
+            <groupId>io.serverlessworkflow</groupId>
+            <artifactId>serverlessworkflow-impl-json</artifactId>
+        </dependency>
         <dependency>
             <groupId>io.serverlessworkflow</groupId>
             <artifactId>serverlessworkflow-fluent-spec</artifactId>

experimental/fluent/func/src/main/java/io/serverlessworkflow/fluent/func/FuncSetTaskBuilder.java

@@ -15,11 +15,28 @@
  */
 package io.serverlessworkflow.fluent.func;
 
+import io.serverlessworkflow.api.types.Set;
+import io.serverlessworkflow.api.types.SetTask;
+import io.serverlessworkflow.api.types.func.MapSetTaskConfiguration;
 import io.serverlessworkflow.fluent.func.spi.ConditionalTaskBuilder;
 import io.serverlessworkflow.fluent.spec.SetTaskBuilder;
+import java.util.Map;
 
 public class FuncSetTaskBuilder extends SetTaskBuilder
     implements ConditionalTaskBuilder<FuncSetTaskBuilder> {
 
-  FuncSetTaskBuilder() {}
+  private final SetTask task;
+
+  FuncSetTaskBuilder() {
+    this.task = new SetTask();
+    this.setTask(task);
+  }
+
+  public FuncSetTaskBuilder expr(Map<String, Object> map) {
+    if (this.task.getSet() == null) {
+      this.task.setSet(new Set());
+    }
+    this.task.getSet().withSetTaskConfiguration(new MapSetTaskConfiguration(map));
+    return this;
+  }
 }

experimental/fluent/func/src/main/java/io/serverlessworkflow/fluent/func/dsl/EmitStep.java

@@ -0,0 +1,51 @@
+/*
+ * 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 io.serverlessworkflow.fluent.func.FuncEmitTaskBuilder;
+import io.serverlessworkflow.fluent.func.FuncTaskItemListBuilder;
+import java.util.Objects;
+import java.util.function.Consumer;
+
+/** Chainable emit step; applies FuncEmitSpec then queued export/when. */
+public final class EmitStep extends Step<EmitStep, FuncEmitTaskBuilder> {
+
+  private final String name; // nullable
+  private final Consumer<FuncEmitTaskBuilder> cfg;
+
+  EmitStep(String name, Consumer<FuncEmitTaskBuilder> cfg) {
+    this.name = name;
+    this.cfg = Objects.requireNonNull(cfg, "cfg");
+  }
+
+  @Override
+  protected void configure(FuncTaskItemListBuilder list, Consumer<FuncEmitTaskBuilder> postApply) {
+    if (name == null) {
+      list.emit(
+          e -> {
+            cfg.accept(e);
+            postApply.accept(e);
+          });
+    } else {
+      list.emit(
+          name,
+          e -> {
+            cfg.accept(e);
+            postApply.accept(e);
+          });
+    }
+  }
+}

experimental/fluent/func/src/main/java/io/serverlessworkflow/fluent/func/dsl/FuncCallStep.java

@@ -0,0 +1,56 @@
+/*
+ * 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 io.serverlessworkflow.fluent.func.FuncCallTaskBuilder;
+import io.serverlessworkflow.fluent.func.FuncTaskItemListBuilder;
+import java.util.function.Consumer;
+import java.util.function.Function;
+
+// FuncCallStep
+public final class FuncCallStep<T, R> extends Step<FuncCallStep<T, R>, FuncCallTaskBuilder> {
+  private final String name; // may be null
+  private final Function<T, R> fn;
+  private final Class<T> argClass;
+
+  FuncCallStep(Function<T, R> fn, Class<T> argClass) {
+    this(null, fn, argClass);
+  }
+
+  FuncCallStep(String name, Function<T, R> fn, Class<T> argClass) {
+    this.name = name;
+    this.fn = fn;
+    this.argClass = argClass;
+  }
+
+  @Override
+  protected void configure(FuncTaskItemListBuilder list, Consumer<FuncCallTaskBuilder> post) {
+    if (name == null) {
+      list.callFn(
+          cb -> {
+            cb.function(fn, argClass);
+            post.accept(cb);
+          });
+    } else {
+      list.callFn(
+          name,
+          cb -> {
+            cb.function(fn, argClass);
+            post.accept(cb);
+          });
+    }
+  }
+}

experimental/fluent/func/src/main/java/io/serverlessworkflow/fluent/func/dsl/FuncDSL.java

@@ -18,7 +18,6 @@
 import io.cloudevents.CloudEventData;
 import io.serverlessworkflow.api.types.FlowDirectiveEnum;
 import io.serverlessworkflow.fluent.func.FuncCallTaskBuilder;
-import io.serverlessworkflow.fluent.func.FuncDoTaskBuilder;
 import io.serverlessworkflow.fluent.func.FuncEmitTaskBuilder;
 import io.serverlessworkflow.fluent.func.FuncSwitchTaskBuilder;
 import io.serverlessworkflow.fluent.func.FuncTaskItemListBuilder;
@@ -28,6 +27,7 @@
 import io.serverlessworkflow.fluent.func.dsl.internal.CommonFuncOps;
 import java.util.Collection;
 import java.util.List;
+import java.util.Map;
 import java.util.Objects;
 import java.util.function.Consumer;
 import java.util.function.Function;
@@ -99,50 +99,90 @@ public static <T> Consumer<FuncEmitTaskBuilder> event(
     return OPS.event(type, function, clazz);
   }
 
+  /** Emit a JSON CloudEvent (PojoCloudEventData) from a POJO payload. */
+  public static <T> Consumer<FuncEmitTaskBuilder> eventJson(String type, Class<T> clazz) {
+    return b -> new FuncEmitSpec().type(type).jsonData(clazz).accept(b);
+  }
+
+  public static <T> Consumer<FuncEmitTaskBuilder> eventBytes(
+      String type, Function<T, byte[]> serializer, Class<T> clazz) {
+    return b -> new FuncEmitSpec().type(type).bytesData(serializer, clazz).accept(b);
+  }
+
+  public static Consumer<FuncEmitTaskBuilder> eventBytesUtf8(String type) {
+    return b -> new FuncEmitSpec().type(type).bytesDataUtf8().accept(b);
+  }
+
   public static FuncPredicateEventConfigurer event(String type) {
     return OPS.event(type);
   }
 
-  public static <T, R> FuncTaskConfigurer function(Function<T, R> fn) {
+  public static <T, R> FuncCallStep<T, R> function(Function<T, R> fn, Class<T> clazz) {
+    return new FuncCallStep<>(fn, clazz);
+  }
+
+  public static <T, R> FuncCallStep<T, R> function(Function<T, R> fn) {
     Class<T> clazz = ReflectionUtils.inferInputType(fn);
-    return list -> list.callFn(f -> f.function(fn, clazz));
+    return new FuncCallStep<>(fn, clazz);
   }
 
-  public static <T, R> FuncTaskConfigurer function(Function<T, R> fn, Class<T> clazz) {
-    return list -> list.callFn(f -> f.function(fn, clazz));
+  public static <T, R> FuncCallStep<T, R> function(String name, Function<T, R> fn) {
+    Class<T> clazz = ReflectionUtils.inferInputType(fn);
+    return new FuncCallStep<>(name, fn, clazz);
   }
 
-  // ------------------  tasks ---------------- //
-  public static Consumer<FuncDoTaskBuilder> doTasks(FuncTaskConfigurer... steps) {
-    final Consumer<FuncTaskItemListBuilder> tasks = tasks(steps);
-    return d -> d.tasks(tasks);
+  public static <T, R> FuncCallStep<T, R> function(String name, Function<T, R> fn, Class<T> clazz) {
+    return new FuncCallStep<>(name, fn, clazz);
   }
 
+  // ------------------  tasks ---------------- //
+
   public static Consumer<FuncTaskItemListBuilder> tasks(FuncTaskConfigurer... steps) {
     Objects.requireNonNull(steps, "Steps in a tasks are required");
     final List<FuncTaskConfigurer> snapshot = List.of(steps.clone());
     return list -> snapshot.forEach(s -> s.accept(list));
   }
 
-  public static FuncTaskConfigurer emit(Consumer<FuncEmitTaskBuilder> emitTask) {
-    return list -> list.emit(emitTask);
+  public static EmitStep emit(Consumer<FuncEmitTaskBuilder> cfg) {
+    return new EmitStep(null, cfg);
+  }
+
+  public static EmitStep emit(String name, Consumer<FuncEmitTaskBuilder> cfg) {
+    return new EmitStep(name, cfg);
+  }
+
+  public static <T> EmitStep emit(String type, Function<T, CloudEventData> fn) {
+    // `event(type, fn)` is your Consumer<FuncEmitTaskBuilder> for EMIT
+    return new EmitStep(null, event(type, fn));
+  }
+
+  public static <T> EmitStep emit(String name, String type, Function<T, CloudEventData> fn) {
+    return new EmitStep(name, event(type, fn));
+  }
+
+  public static <T> EmitStep emit(
+      String name, String type, Function<T, byte[]> serializer, Class<T> clazz) {
+    return new EmitStep(name, eventBytes(type, serializer, clazz));
+  }
+
+  public static <T> EmitStep emit(String type, Function<T, byte[]> serializer, Class<T> clazz) {
+    return new EmitStep(null, eventBytes(type, serializer, clazz));
   }
 
-  public static <T> FuncTaskConfigurer emit(String type, Function<T, CloudEventData> fn) {
-    return list -> list.emit(event(type, fn));
+  public static <T> EmitStep emitJson(String type, Class<T> clazz) {
+    return new EmitStep(null, eventJson(type, clazz));
   }
 
-  public static <T> FuncTaskConfigurer emit(
-      String name, String type, Function<T, CloudEventData> fn) {
-    return list -> list.emit(name, event(type, fn));
+  public static <T> EmitStep emitJson(String name, String type, Class<T> clazz) {
+    return new EmitStep(name, eventJson(type, clazz));
   }
 
-  public static FuncTaskConfigurer listen(FuncListenSpec listen) {
-    return list -> list.listen(listen);
+  public static ListenStep listen(FuncListenSpec spec) {
+    return new ListenStep(null, spec);
   }
 
-  public static FuncTaskConfigurer listen(String name, FuncListenSpec listen) {
-    return list -> list.listen(name, listen);
+  public static ListenStep listen(String name, FuncListenSpec spec) {
+    return new ListenStep(name, spec);
   }
 
   public static FuncTaskConfigurer switchCase(
@@ -176,6 +216,12 @@ public static <T> FuncTaskConfigurer switchWhenOrElse(
         list.switchCase(FuncDSL.cases(caseOf(pred).then(thenTask), caseDefault(otherwise)));
   }
 
+  public static <T> FuncTaskConfigurer switchWhenOrElse(
+      Predicate<T> pred, String thenTask, String otherwiseTask) {
+    return list ->
+        list.switchCase(FuncDSL.cases(caseOf(pred).then(thenTask), caseDefault(otherwiseTask)));
+  }
+
   public static <T> FuncTaskConfigurer forEach(
       Function<T, Collection<?>> collection, Consumer<FuncTaskItemListBuilder> body) {
     return list -> list.forEach(j -> j.collection(collection).tasks(body));
@@ -192,4 +238,12 @@ public static <T> FuncTaskConfigurer forEach(
       List<T> collection, Consumer<FuncTaskItemListBuilder> body) {
     return list -> list.forEach(j -> j.collection(ctx -> collection).tasks(body));
   }
+
+  public static FuncTaskConfigurer set(String expr) {
+    return list -> list.set(expr);
+  }
+
+  public static FuncTaskConfigurer set(Map<String, Object> map) {
+    return list -> list.set(s -> s.expr(map));
+  }
 }

experimental/fluent/func/src/main/java/io/serverlessworkflow/fluent/func/dsl/FuncEventFilterSpec.java

@@ -16,9 +16,13 @@
 package io.serverlessworkflow.fluent.func.dsl;
 
 import io.cloudevents.CloudEventData;
+import io.cloudevents.core.data.BytesCloudEventData;
+import io.cloudevents.core.data.PojoCloudEventData;
 import io.serverlessworkflow.api.types.func.EventDataFunction;
 import io.serverlessworkflow.fluent.func.FuncEventPropertiesBuilder;
 import io.serverlessworkflow.fluent.spec.dsl.EventFilterSpec;
+import io.serverlessworkflow.impl.jackson.JsonUtils;
+import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.function.Function;
 
@@ -36,9 +40,31 @@ public <T> SELF jsonData(Function<T, CloudEventData> function) {
     return JSON();
   }
 
+  /** Sets the event data and the contentType to `application/octet-stream` */
+  public <T> SELF bytesData(Function<T, byte[]> serializer, Class<T> clazz) {
+    addStep(e -> e.data(payload -> BytesCloudEventData.wrap(serializer.apply(payload)), clazz));
+    return OCTET_STREAM();
+  }
+
+  public SELF bytesDataUtf8() {
+    return bytesData((String s) -> s.getBytes(StandardCharsets.UTF_8), String.class);
+  }
+
   /** Sets the event data and the contentType to `application/json` */
   public <T> SELF jsonData(Function<T, CloudEventData> function, Class<T> clazz) {
     addStep(e -> e.data(new EventDataFunction().withFunction(function, clazz)));
     return JSON();
   }
+
+  /** JSON with default mapper (PojoCloudEventData + application/json). */
+  public <T> SELF jsonData(Class<T> clazz) {
+    addStep(
+        e ->
+            e.data(
+                payload ->
+                    PojoCloudEventData.wrap(
+                        payload, p -> JsonUtils.mapper().writeValueAsString(p).getBytes()),
+                clazz));
+    return JSON();
+  }
 }

experimental/fluent/func/src/main/java/io/serverlessworkflow/fluent/func/dsl/ListenStep.java

@@ -0,0 +1,51 @@
+/*
+ * 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 io.serverlessworkflow.fluent.func.FuncListenTaskBuilder;
+import io.serverlessworkflow.fluent.func.FuncTaskItemListBuilder;
+import java.util.function.Consumer;
+
+/** Chainable listen step; applies FuncListenSpec then queued export/when. */
+public final class ListenStep extends Step<ListenStep, FuncListenTaskBuilder> {
+
+  private final String name; // nullable
+  private final FuncListenSpec spec;
+
+  ListenStep(String name, FuncListenSpec spec) {
+    this.name = name;
+    this.spec = spec;
+  }
+
+  @Override
+  protected void configure(
+      FuncTaskItemListBuilder list, Consumer<FuncListenTaskBuilder> postApply) {
+    if (name == null) {
+      list.listen(
+          lb -> {
+            spec.accept(lb);
+            postApply.accept(lb);
+          });
+    } else {
+      list.listen(
+          name,
+          lb -> {
+            spec.accept(lb);
+            postApply.accept(lb);
+          });
+    }
+  }
+}