Implement asyncapi DSL

fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/BaseTaskItemListBuilder.java

@@ -46,6 +46,7 @@ public abstract class BaseTaskItemListBuilder<SELF extends BaseTaskItemListBuild
   protected final String TYPE_HTTP = "http";
   protected final String TYPE_OPENAPI = "openapi";
   protected final String TYPE_GRPC = "grpc";
+  protected final String TYPE_ASYNCAPI = "asyncapi";
   protected final String TYPE_WORKFLOW = "workflow";
 
   private final List<TaskItem> list;

fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/BaseWorkflowBuilder.java

@@ -136,7 +136,11 @@ private SELF appendDo(Consumer<DBuilder> configurer) {
     configurer.accept(doBuilder);
 
     final List<TaskItem> newItems = doBuilder.build().getDo();
-    if (newItems == null || newItems.isEmpty()) return self();
+    if (newItems == null || newItems.isEmpty()) {
+      throw new IllegalStateException(
+          "Task list must contain at least one task. "
+              + "Use .tasks(d -> d.set(...)) or similar to define tasks.");
+    }
 
     final List<TaskItem> merged =
         new ArrayList<>(this.workflow.getDo() != null ? this.workflow.getDo() : List.of());

fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/CallAsyncAPITaskBuilder.java

@@ -0,0 +1,35 @@
+/*
+ * 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.spec;
+
+import io.serverlessworkflow.api.types.AsyncApiArguments;
+import io.serverlessworkflow.api.types.CallAsyncAPI;
+import io.serverlessworkflow.fluent.spec.spi.CallAsyncAPITaskFluent;
+
+public class CallAsyncAPITaskBuilder extends TaskBaseBuilder<CallAsyncAPITaskBuilder>
+    implements CallAsyncAPITaskFluent<CallAsyncAPITaskBuilder> {
+
+  CallAsyncAPITaskBuilder() {
+    final CallAsyncAPI callAsyncAPI = new CallAsyncAPI();
+    callAsyncAPI.setWith(new AsyncApiArguments());
+    super.setTask(callAsyncAPI);
+  }
+
+  @Override
+  public CallAsyncAPITaskBuilder self() {
+    return this;
+  }
+}

fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/DoTaskBuilder.java

@@ -104,6 +104,12 @@ public DoTaskBuilder openapi(String name, Consumer<CallOpenAPITaskBuilder> items
     return this;
   }
 
+  @Override
+  public DoTaskBuilder asyncapi(String name, Consumer<CallAsyncAPITaskBuilder> itemsConfigurer) {
+    this.listBuilder().asyncapi(name, itemsConfigurer);
+    return this;
+  }
+
   @Override
   public DoTaskBuilder grpc(String name, Consumer<CallGrpcTaskBuilder> itemsConfigurer) {
     this.listBuilder().grpc(name, itemsConfigurer);

fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/TaskItemListBuilder.java

@@ -154,6 +154,22 @@ public TaskItemListBuilder openapi(
     return addTaskItem(new TaskItem(name, task));
   }
 
+  @Override
+  public TaskItemListBuilder asyncapi(
+      String name, Consumer<CallAsyncAPITaskBuilder> itemsConfigurer) {
+    name = defaultNameAndRequireConfig(name, itemsConfigurer, TYPE_ASYNCAPI);
+
+    final CallAsyncAPITaskBuilder callAsyncAPIBuilder = new CallAsyncAPITaskBuilder();
+    itemsConfigurer.accept(callAsyncAPIBuilder);
+
+    final CallTask callTask = new CallTask();
+    callTask.setCallAsyncAPI(callAsyncAPIBuilder.build());
+    final Task task = new Task();
+    task.setCallTask(callTask);
+
+    return addTaskItem(new TaskItem(name, task));
+  }
+
   @Override
   public TaskItemListBuilder grpc(String name, Consumer<CallGrpcTaskBuilder> itemsConfigurer) {
     name = defaultNameAndRequireConfig(name, itemsConfigurer, TYPE_GRPC);

fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/configurers/CallAsyncAPIConfigurer.java

@@ -0,0 +1,22 @@
+/*
+ * 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.spec.configurers;
+
+import io.serverlessworkflow.fluent.spec.CallAsyncAPITaskBuilder;
+import java.util.function.Consumer;
+
+@FunctionalInterface
+public interface CallAsyncAPIConfigurer extends Consumer<CallAsyncAPITaskBuilder> {}

fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/dsl/CallAsyncAPISpec.java

@@ -0,0 +1,137 @@
+/*
+ * 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.spec.dsl;
+
+import io.serverlessworkflow.api.types.AsyncApiArguments;
+import io.serverlessworkflow.fluent.spec.CallAsyncAPITaskBuilder;
+import io.serverlessworkflow.fluent.spec.SubscriptionIteratorBuilder;
+import io.serverlessworkflow.fluent.spec.TaskItemListBuilder;
+import io.serverlessworkflow.fluent.spec.configurers.AuthenticationConfigurer;
+import io.serverlessworkflow.fluent.spec.configurers.CallAsyncAPIConfigurer;
+import io.serverlessworkflow.fluent.spec.spi.CallAsyncAPITaskFluent;
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Consumer;
+
+public final class CallAsyncAPISpec implements CallAsyncAPIConfigurer {
+
+  private final List<Consumer<CallAsyncAPITaskFluent<?>>> steps = new ArrayList<>();
+
+  public CallAsyncAPISpec document(String uri) {
+    steps.add(b -> b.document(uri));
+    return this;
+  }
+
+  public CallAsyncAPISpec document(String uri, AuthenticationConfigurer authenticationConfigurer) {
+    steps.add(b -> b.document(uri, authenticationConfigurer));
+    return this;
+  }
+
+  public CallAsyncAPISpec document(URI uri) {
+    steps.add(b -> b.document(uri));
+    return this;
+  }
+
+  public CallAsyncAPISpec document(URI uri, AuthenticationConfigurer authenticationConfigurer) {
+    steps.add(b -> b.document(uri, authenticationConfigurer));
+    return this;
+  }
+
+  public CallAsyncAPISpec channel(String channel) {
+    steps.add(b -> b.channel(channel));
+    return this;
+  }
+
+  public CallAsyncAPISpec operation(String operation) {
+    steps.add(b -> b.operation(operation));
+    return this;
+  }
+
+  public CallAsyncAPISpec server(String name) {
+    steps.add(b -> b.server(name));
+    return this;
+  }
+
+  public CallAsyncAPISpec server(String name, Map<String, Object> variables) {
+    steps.add(b -> b.server(name, variables));
+    return this;
+  }
+
+  public CallAsyncAPISpec protocol(AsyncApiArguments.AsyncApiProtocol protocol) {
+    steps.add(b -> b.protocol(protocol));
+    return this;
+  }
+
+  public CallAsyncAPISpec message(Map<String, Object> payload) {
+    steps.add(b -> b.message(payload));
+    return this;
+  }
+
+  public CallAsyncAPISpec message(Map<String, Object> payload, Map<String, Object> headers) {
+    steps.add(b -> b.message(payload, headers));
+    return this;
+  }
+
+  public CallAsyncAPISpec payload(Map<String, Object> payload) {
+    steps.add(b -> b.payload(payload));
+    return this;
+  }
+
+  public CallAsyncAPISpec headers(Map<String, Object> headers) {
+    steps.add(b -> b.headers(headers));
+    return this;
+  }
+
+  public CallAsyncAPISpec consumeAmount(int amount) {
+    steps.add(b -> b.consumeAmount(amount));
+    return this;
+  }
+
+  public CallAsyncAPISpec consumeWhile(String expression) {
+    steps.add(b -> b.consumeWhile(expression));
+    return this;
+  }
+
+  public CallAsyncAPISpec consumeUntil(String expression) {
+    steps.add(b -> b.consumeUntil(expression));
+    return this;
+  }
+
+  public CallAsyncAPISpec filter(String filterExpression) {
+    steps.add(b -> b.filter(filterExpression));
+    return this;
+  }
+
+  public CallAsyncAPISpec subscription(
+      Consumer<SubscriptionIteratorBuilder<TaskItemListBuilder>> foreachConfigurer) {
+    steps.add(b -> b.subscription(foreachConfigurer));
+    return this;
+  }
+
+  public CallAsyncAPISpec authentication(AuthenticationConfigurer authenticationConfigurer) {
+    steps.add(b -> b.authentication(authenticationConfigurer));
+    return this;
+  }
+
+  @Override
+  public void accept(CallAsyncAPITaskBuilder builder) {
+    for (var s : steps) {
+      s.accept(builder);
+    }
+  }
+}

fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/dsl/DSL.java

@@ -28,6 +28,7 @@
 import io.serverlessworkflow.fluent.spec.TimeoutBuilder;
 import io.serverlessworkflow.fluent.spec.TryTaskBuilder;
 import io.serverlessworkflow.fluent.spec.configurers.AuthenticationConfigurer;
+import io.serverlessworkflow.fluent.spec.configurers.CallAsyncAPIConfigurer;
 import io.serverlessworkflow.fluent.spec.configurers.CallGrpcConfigurer;
 import io.serverlessworkflow.fluent.spec.configurers.CallHttpConfigurer;
 import io.serverlessworkflow.fluent.spec.configurers.CallOpenAPIConfigurer;
@@ -114,6 +115,28 @@ public static CallGrpcSpec grpc() {
     return new CallGrpcSpec();
   }
 
+  /**
+   * Create a new AsyncAPI call specification to be used with {@link #call(CallAsyncAPIConfigurer)}.
+   *
+   * <p>Typical usage:
+   *
+   * <pre>{@code
+   * tasks(
+   *   call(
+   *     asyncapi()
+   *       .document("http://acme.org/asyncapi.yaml")
+   *       .operation("greet")
+   *       .message(Map.of("greeting", "hello"))
+   *   )
+   * );
+   * }</pre>
+   *
+   * @return a new {@link CallAsyncAPISpec} instance
+   */
+  public static CallAsyncAPISpec asyncapi() {
+    return new CallAsyncAPISpec();
+  }
+
   public static WorkflowSpec workflow(String namespace, String name, String version) {
     return new WorkflowSpec().namespace(namespace).name(name).version(version);
   }
@@ -760,6 +783,27 @@ public static TasksConfigurer call(String name, CallOpenAPIConfigurer configurer
     return list -> list.openapi(name, configurer);
   }
 
+  /**
+   * Create a {@link TasksConfigurer} that adds an AsyncAPI call task.
+   *
+   * @param configurer AsyncAPI configurer
+   * @return a {@link TasksConfigurer} that adds a CallAsyncAPI task
+   */
+  public static TasksConfigurer call(CallAsyncAPIConfigurer configurer) {
+    return list -> list.asyncapi(configurer);
+  }
+
+  /**
+   * Create a {@link TasksConfigurer} that adds an AsyncAPI call task with an explicit name.
+   *
+   * @param name the task name
+   * @param configurer AsyncAPI configurer
+   * @return a {@link TasksConfigurer} that adds a CallAsyncAPI task
+   */
+  public static TasksConfigurer call(String name, CallAsyncAPIConfigurer configurer) {
+    return list -> list.asyncapi(name, configurer);
+  }
+
   public static TasksConfigurer call(CallGrpcConfigurer configurer) {
     return list -> list.grpc(configurer);
   }

fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/spi/CallAsyncAPIFluent.java

@@ -0,0 +1,28 @@
+/*
+ * 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.spec.spi;
+
+import io.serverlessworkflow.fluent.spec.TaskBaseBuilder;
+import java.util.function.Consumer;
+
+public interface CallAsyncAPIFluent<SELF extends TaskBaseBuilder<SELF>, LIST> {
+
+  LIST asyncapi(String name, Consumer<SELF> itemsConfigurer);
+
+  default LIST asyncapi(Consumer<SELF> itemsConfigurer) {
+    return this.asyncapi(null, itemsConfigurer);
+  }
+}