Implement asyncapi call + asyncapi DSL

[mcruzdev]

Many thanks for submitting your Pull Request ❤️!

What this PR does / why we need it:

Special notes for reviewers:

Additional information (if needed):

Closes #1607, #1618

[Copilot]

Pull request overview

Adds AsyncAPI support to the Java workflow runtime by introducing a new impl/asyncapi module that implements the call: asyncapi task (publish + subscribe) and wires it into the runtime via SPI, along with test fixtures to validate behavior.

Changes:

• Introduces impl/asyncapi module with AsyncAPIExecutor/builder, lightweight AsyncAPI document model + reader, and a pluggable AsyncApiChannelProvider SPI.
• Adds JUnit tests plus AsyncAPI spec/workflow YAML fixtures for publish and multiple subscribe policies.
• Updates Maven module wiring and test module dependencies to include the new asyncapi implementation.

Reviewed changes

Copilot reviewed 20 out of 20 changed files in this pull request and generated 4 comments.

Show a summary per file

File	Description
impl/test/src/test/resources/workflows-samples/asyncapi/asyncapi-subscribe-until.yaml	Adds a workflow fixture for subscribe + until consumption policy.
impl/test/src/test/resources/workflows-samples/asyncapi/asyncapi-subscribe-foreach.yaml	Adds a workflow fixture for subscribe + foreach per-message processing.
impl/test/src/test/resources/workflows-samples/asyncapi/asyncapi-subscribe-filter.yaml	Adds a workflow fixture for subscribe + message filtering.
impl/test/src/test/resources/workflows-samples/asyncapi/asyncapi-subscribe-amount.yaml	Adds a workflow fixture for subscribe + amount consumption policy.
impl/test/src/test/resources/workflows-samples/asyncapi/asyncapi-publish.yaml	Adds a workflow fixture for publish with payload + headers.
impl/test/src/test/resources/schema/asyncapi/asyncapi.yaml	Adds an AsyncAPI 3.0 test document fixture used by tests.
impl/test/src/test/java/io/serverlessworkflow/impl/test/AsyncAPITest.java	Adds integration-style tests using an in-memory provider and MockWebServer-hosted AsyncAPI spec.
impl/test/pom.xml	Adds test-scope dependency on serverlessworkflow-impl-asyncapi.
impl/pom.xml	Adds serverlessworkflow-impl-asyncapi to dependency management and module list.
impl/asyncapi/src/main/resources/META-INF/services/io.serverlessworkflow.impl.executors.CallableTaskBuilder	Registers AsyncAPIExecutorBuilder via Java SPI.
impl/asyncapi/src/main/java/io/serverlessworkflow/impl/executors/asyncapi/UnifiedAsyncAPI.java	Adds a minimal unified AsyncAPI document model for server/channel/operation resolution.
impl/asyncapi/src/main/java/io/serverlessworkflow/impl/executors/asyncapi/AsyncApiSubscriptionHandle.java	Adds subscription lifecycle abstraction (unsubscribe + closed future).
impl/asyncapi/src/main/java/io/serverlessworkflow/impl/executors/asyncapi/AsyncAPIReader.java	Adds AsyncAPI document reader using workflow format mappers (YAML/JSON).
impl/asyncapi/src/main/java/io/serverlessworkflow/impl/executors/asyncapi/AsyncApiInboundMessage.java	Adds inbound message envelope (payload/headers/correlationId) with null-safe defaults.
impl/asyncapi/src/main/java/io/serverlessworkflow/impl/executors/asyncapi/AsyncAPIExecutorBuilder.java	Builds executor configs (payload/header resolvers, predicates, foreach executor, timeout resolver).
impl/asyncapi/src/main/java/io/serverlessworkflow/impl/executors/asyncapi/AsyncAPIExecutor.java	Implements publish/subscribe execution, document/server/channel resolution, and consumption policy evaluation.
impl/asyncapi/src/main/java/io/serverlessworkflow/impl/executors/asyncapi/AsyncApiChannelProvider.java	Adds transport SPI for publish and subscribe operations.
impl/asyncapi/src/main/java/io/serverlessworkflow/impl/executors/asyncapi/AsyncApiChannelInfo.java	Adds resolved channel metadata passed to providers (URI, protocol, operation, auth token).
impl/asyncapi/pom.xml	Introduces the new serverlessworkflow-impl-asyncapi Maven module.
asyncapi-call-plan.md	Adds an implementation plan / design notes for the AsyncAPI call feature.

Suppressed comments (3)

impl/asyncapi/src/main/java/io/serverlessworkflow/impl/executors/asyncapi/AsyncAPIExecutor.java:246

• AsyncApiSubscriptionHandle.closed() is never observed, so if the provider closes the subscription early (or fails) the task future may never complete (and can hang indefinitely unless a consume timeout is configured). Wire handle.closed() into the task result so provider-side shutdown/errors terminate the task deterministically.

    result.whenComplete((r, ex) -> handle.unsubscribe());

impl/asyncapi/src/main/java/io/serverlessworkflow/impl/executors/asyncapi/AsyncAPIExecutor.java:279

• processTaskList(...).join() blocks the provider callback thread (and it’s currently executed while holding the synchronized (collection) lock). This can severely limit throughput and can deadlock/starve if the subscription callback is invoked on the same executor used to run workflow tasks. Prefer a non-blocking FIFO chain (e.g., keep a CompletableFuture<Void> tail and thenCompose per message) and avoid holding locks while waiting for workflow task completion.

    if (subscribeConfig.foreachExecutor() != null) {
      taskContext.variables().put(subscribeConfig.foreachItem(), messageModel);
      taskContext.variables().put(subscribeConfig.foreachAt(), collection.size());
      return TaskExecutorHelper.processTaskList(
              subscribeConfig.foreachExecutor(),
              workflowContext,
              Optional.of(taskContext),
              messageModel)
          .join();

impl/asyncapi/src/main/java/io/serverlessworkflow/impl/executors/asyncapi/AsyncAPIExecutor.java:261

• The executor implements consume.for timeout behavior (graceful completion with partial results), but there’s no test covering this path yet. Adding a test + workflow fixture that sets subscription.consume.for and verifies the result completes with a partial collection would protect this behavior from regressions.

    subscribeConfig
        .consumeTimeout()
        .ifPresent(
            resolver -> {
              Duration duration = resolver.apply(workflowContext, taskContext, input);
              CompletableFuture.delayedExecutor(duration.toMillis(), TimeUnit.MILLISECONDS)
                  .execute(
                      () -> {
                        synchronized (collection) {
                          if (!result.isDone()) {
                            result.complete(collection);
                          }
                        }
                      });
            });

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.