Add defaultEventSource to WorkflowApplication

impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowApplication.java

@@ -30,6 +30,7 @@
 import io.serverlessworkflow.impl.config.SystemPropertyConfigManager;
 import io.serverlessworkflow.impl.events.CloudEventPredicateFactory;
 import io.serverlessworkflow.impl.events.DefaultCloudEventPredicateFactory;
+import io.serverlessworkflow.impl.events.EmitSourceResolver;
 import io.serverlessworkflow.impl.events.EventConsumer;
 import io.serverlessworkflow.impl.events.EventPublisher;
 import io.serverlessworkflow.impl.events.InMemoryEvents;
@@ -105,6 +106,7 @@ public class WorkflowApplication implements AutoCloseable {
   private final Optional<URITemplateResolver> templateResolver;
   private final Optional<FunctionReader> functionReader;
   private final URI defaultCatalogURI;
+  private final WorkflowValueResolver<URI> defaultEventSource;
   private final Collection<CallableTaskProxyBuilder> callableProxyBuilders;
   private final CloudEventPredicateFactory cloudEventPredicateFactory;
   private final AllStrategyCorrelationInfoFactory allStrategyCorrelationInfoFactory;
@@ -138,6 +140,7 @@ private WorkflowApplication(Builder builder) {
     this.templateResolver = builder.templateResolver;
     this.functionReader = builder.functionReader;
     this.defaultCatalogURI = builder.defaultCatalogURI;
+    this.defaultEventSource = builder.defaultEventSource;
     this.id = builder.id;
     this.callableProxyBuilders = builder.callableProxyBuilders;
     this.cloudEventPredicateFactory = builder.cloudEventPredicateFactory;
@@ -263,6 +266,7 @@ public SchemaValidator getValidator(SchemaInline inline) {
     private Optional<URITemplateResolver> templateResolver;
     private Optional<FunctionReader> functionReader;
     private URI defaultCatalogURI;
+    private WorkflowValueResolver<URI> defaultEventSource = new EmitSourceResolver();
     private CloudEventPredicateFactory cloudEventPredicateFactory;
     private AllStrategyCorrelationInfoFactory allStrategyCorrelationInfoFactory;
     private WorkflowLifeCycleCloudEventFactory lifeCycleCloudEventFactory;
@@ -404,6 +408,19 @@ public Builder withDefaultCatalogURI(URI defaultCatalogURI) {
       return this;
     }
 
+    public Builder withDefaultEventSource(String defaultEventSource) {
+      return withDefaultEventSource(URI.create(defaultEventSource));
+    }
+
+    public Builder withDefaultEventSource(URI defaultEventSource) {
+      return withDefaultEventSource((workflow, task, model) -> defaultEventSource);
+    }
+
+    public Builder withDefaultEventSource(WorkflowValueResolver<URI> defaultEventSource) {
+      this.defaultEventSource = defaultEventSource;
+      return this;
+    }
+
     public Builder withCloudEventPredicateFactory(
         CloudEventPredicateFactory cloudEventPredicateFactory) {
       this.cloudEventPredicateFactory = cloudEventPredicateFactory;
@@ -639,6 +656,10 @@ public URI defaultCatalogURI() {
     return defaultCatalogURI;
   }
 
+  public WorkflowValueResolver<URI> defaultEventSource() {
+    return defaultEventSource;
+  }
+
   public String id() {
     return id;
   }

impl/core/src/main/java/io/serverlessworkflow/impl/events/CloudEventUtils.java

@@ -45,6 +45,6 @@ public static Map<String, Object> extensions(CloudEvent event) {
   }
 
   public static URI source() {
-    return URI.create("reference-impl");
+    return URI.create("io.serverlessworkflow.sdk-java");
   }
 }

impl/core/src/main/java/io/serverlessworkflow/impl/events/EmitSourceResolver.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.impl.events;
+
+import io.serverlessworkflow.impl.TaskContext;
+import io.serverlessworkflow.impl.WorkflowContext;
+import io.serverlessworkflow.impl.WorkflowDefinitionId;
+import io.serverlessworkflow.impl.WorkflowModel;
+import io.serverlessworkflow.impl.WorkflowValueResolver;
+import java.net.URI;
+
+public class EmitSourceResolver implements WorkflowValueResolver<URI> {
+
+  private static final String SEP = "/";
+  private static final String VER_SEP = ":";
+
+  @Override
+  public URI apply(WorkflowContext workflow, TaskContext task, WorkflowModel model) {
+    WorkflowDefinitionId id = workflow.definition().id();
+    return URI.create(id.namespace() + SEP + id.name() + VER_SEP + id.version());
+  }
+}

impl/core/src/main/java/io/serverlessworkflow/impl/executors/EmitExecutor.java

@@ -100,17 +100,24 @@ private CloudEvent buildCloudEvent(WorkflowContext workflow, TaskContext taskCon
             .sourceFilter()
             .map(filter -> filter.apply(workflow, taskContext, taskContext.input()))
             .map(URI::create)
-            .orElse(CloudEventUtils.source()));
+            .orElseGet(
+                () ->
+                    workflow
+                        .definition()
+                        .application()
+                        .defaultEventSource()
+                        .apply(workflow, taskContext, taskContext.input())));
     ceBuilder.withType(
         props
             .typeFilter()
             .map(filter -> filter.apply(workflow, taskContext, taskContext.input()))
             .orElseThrow(
                 () -> new IllegalArgumentException("Type is required for emitting events")));
-    props
-        .timeFilter()
-        .map(filter -> filter.apply(workflow, taskContext, taskContext.input()))
-        .ifPresent(value -> ceBuilder.withTime(value));
+    ceBuilder.withTime(
+        props
+            .timeFilter()
+            .map(filter -> filter.apply(workflow, taskContext, taskContext.input()))
+            .orElseGet(OffsetDateTime::now));
     props
         .subjectFilter()
         .map(filter -> filter.apply(workflow, taskContext, taskContext.input()))

impl/test/src/test/java/io/serverlessworkflow/impl/test/EmitExecutorTest.java

@@ -0,0 +1,198 @@
+/*
+ * 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.impl.test;
+
+import static io.serverlessworkflow.api.WorkflowReader.readWorkflowFromClasspath;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import io.cloudevents.CloudEvent;
+import io.serverlessworkflow.api.types.Workflow;
+import io.serverlessworkflow.fluent.spec.EventPropertiesBuilder;
+import io.serverlessworkflow.fluent.spec.WorkflowBuilder;
+import io.serverlessworkflow.impl.WorkflowApplication;
+import io.serverlessworkflow.impl.events.InMemoryEvents;
+import java.io.IOException;
+import java.net.URI;
+import java.time.OffsetDateTime;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.function.Consumer;
+import org.junit.jupiter.api.Test;
+
+class EmitExecutorTest {
+
+  private static List<CloudEvent> runWorkflow(Workflow wf, String eventType) {
+    List<CloudEvent> events = Collections.synchronizedList(new ArrayList<>());
+    InMemoryEvents broker = new InMemoryEvents();
+    broker.register(eventType, events::add);
+    WorkflowApplication app =
+        WorkflowApplication.builder().withEventConsumer(broker).withEventPublisher(broker).build();
+    try {
+      app.workflowDefinition(wf).instance().start().join();
+    } finally {
+      app.close();
+    }
+    return events;
+  }
+
+  private static Workflow emitWorkflow(String name, Consumer<EventPropertiesBuilder> props) {
+    return WorkflowBuilder.workflow(name, "test", "0.1.0")
+        .tasks(t -> t.emit("emitEvent", etb -> etb.event(props::accept)))
+        .build();
+  }
+
+  private static Workflow emitWorkflow(String name, String type) {
+    return emitWorkflow(name, epb -> epb.type(type));
+  }
+
+  @Test
+  void whenMissingType_throwsIllegalArgumentException() throws IOException {
+    Workflow wf =
+        WorkflowBuilder.workflow("emit-no-type", "test", "0.1.0")
+            .tasks(t -> t.emit("emitEvent", etb -> etb.event(epb -> {})))
+            .build();
+
+    assertThatThrownBy(() -> runWorkflow(wf, "any"))
+        .isInstanceOf(java.util.concurrent.CompletionException.class)
+        .hasCauseInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining("Type is required for emitting events");
+  }
+
+  @Test
+  void whenTypeOnly_sourceUsesDefinitionIdFQN() throws IOException {
+    Workflow wf = emitWorkflow("emit-source-resolver", "org.example.event");
+    List<CloudEvent> events = runWorkflow(wf, "org.example.event");
+
+    assertThat(events).hasSize(1);
+    CloudEvent ev = events.get(0);
+    assertThat(ev.getType()).isEqualTo("org.example.event");
+    assertThat(ev.getSource()).isEqualTo(URI.create("test/emit-source-resolver:0.1.0"));
+  }
+
+  @Test
+  void whenAppHasDefaultEventSource_itOverridesDefinitionIdFQN() throws IOException {
+    Workflow wf = emitWorkflow("emit-app-default", "org.example.event");
+
+    List<CloudEvent> events = Collections.synchronizedList(new ArrayList<>());
+    InMemoryEvents broker = new InMemoryEvents();
+    broker.register("org.example.event", events::add);
+    WorkflowApplication app =
+        WorkflowApplication.builder()
+            .withDefaultEventSource("my.custom.source")
+            .withEventConsumer(broker)
+            .withEventPublisher(broker)
+            .build();
+    app.workflowDefinition(wf).instance().start().join();
+    app.close();
+
+    assertThat(events).hasSize(1);
+    CloudEvent ev = events.get(0);
+    assertThat(ev.getSource()).isEqualTo(URI.create("my.custom.source"));
+  }
+
+  @Test
+  void whenExplicitSourceInEventDefinition_itTakesPrecedenceOverAllDefaults() throws IOException {
+    Workflow wf =
+        WorkflowBuilder.workflow("emit-explicit-source", "my.ns", "1.2.3")
+            .tasks(
+                t ->
+                    t.emit(
+                        "emitEvent",
+                        etb ->
+                            etb.event(
+                                epb -> {
+                                  epb.type("org.example.event");
+                                  epb.source("explicit/source");
+                                })))
+            .build();
+
+    List<CloudEvent> events = Collections.synchronizedList(new ArrayList<>());
+    InMemoryEvents broker = new InMemoryEvents();
+    broker.register("org.example.event", events::add);
+    WorkflowApplication app =
+        WorkflowApplication.builder()
+            .withDefaultEventSource("app.source")
+            .withEventConsumer(broker)
+            .withEventPublisher(broker)
+            .build();
+    app.workflowDefinition(wf).instance().start().join();
+    app.close();
+
+    assertThat(events).hasSize(1);
+    CloudEvent ev = events.get(0);
+    assertThat(ev.getSource()).isEqualTo(URI.create("explicit/source"));
+  }
+
+  @Test
+  void whenDefaultEventSourceIsResolver_uriIsResolvedFromWorkflowContext() throws IOException {
+    Workflow wf = emitWorkflow("emit-resolver-default", "org.example.event");
+
+    List<CloudEvent> events = Collections.synchronizedList(new ArrayList<>());
+    InMemoryEvents broker = new InMemoryEvents();
+    broker.register("org.example.event", events::add);
+    WorkflowApplication app =
+        WorkflowApplication.builder()
+            .withDefaultEventSource(
+                (workflow, task, model) -> URI.create("apps/" + workflow.definition().id().name()))
+            .withEventConsumer(broker)
+            .withEventPublisher(broker)
+            .build();
+    app.workflowDefinition(wf).instance().start().join();
+    app.close();
+
+    assertThat(events).hasSize(1);
+    CloudEvent ev = events.get(0);
+    assertThat(ev.getSource()).isEqualTo(URI.create("apps/emit-resolver-default"));
+  }
+
+  @Test
+  void whenWorkflowIsLoadedFromYaml_sourceUsesDefinitionIdFQN() throws IOException {
+    Workflow wf = readWorkflowFromClasspath("workflows-samples/emit-default-source.yaml");
+    List<CloudEvent> events = runWorkflow(wf, "com.test.default.source");
+
+    assertThat(events).hasSize(1);
+    CloudEvent ev = events.get(0);
+    assertThat(ev.getType()).isEqualTo("com.test.default.source");
+    assertThat(ev.getSource()).isEqualTo(URI.create("test/emit-default-source:0.1.0"));
+    assertThat(ev.getTime()).isNotNull();
+    assertThat(ev.getId()).isNotNull().isNotEmpty();
+  }
+
+  @Test
+  void whenTimeIsNotSpecified_defaultsToNow() throws IOException {
+    Workflow wf = emitWorkflow("emit-time-default", "org.example.event");
+    List<CloudEvent> events = runWorkflow(wf, "org.example.event");
+
+    assertThat(events).hasSize(1);
+    CloudEvent ev = events.get(0);
+    OffsetDateTime now = OffsetDateTime.now();
+    assertThat(ev.getTime()).isNotNull();
+    assertThat(ev.getTime()).isBeforeOrEqualTo(now);
+    assertThat(ev.getTime()).isAfter(now.minusSeconds(30));
+  }
+
+  @Test
+  void whenIdIsNotSpecified_autoGeneratedIdIsSet() throws IOException {
+    Workflow wf = emitWorkflow("emit-id-default", "org.example.event");
+    List<CloudEvent> events = runWorkflow(wf, "org.example.event");
+
+    assertThat(events).hasSize(1);
+    CloudEvent ev = events.get(0);
+    assertThat(ev.getId()).isNotNull().isNotEmpty();
+  }
+}

impl/test/src/test/resources/workflows-samples/emit-default-source.yaml

@@ -0,0 +1,13 @@
+document:
+  dsl: '1.0.0-alpha5'
+  namespace: test
+  name: emit-default-source
+  version: '0.1.0'
+do:
+  - emitEvent:
+      emit:
+        event:
+          with:
+            type: com.test.default.source
+            data:
+              message: hello