From 37dd274c526cf6b8b9c446ad18f13ab9aa44e54a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Matheus=20Andr=C3=A9?= Date: Wed, 22 Jul 2026 15:22:54 -0300 Subject: [PATCH] Add defaultEventSource to WorkflowApplication MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Matheus André --- .../impl/WorkflowApplication.java | 21 ++ .../impl/events/CloudEventUtils.java | 2 +- .../impl/events/EmitSourceResolver.java | 35 ++++ .../impl/executors/EmitExecutor.java | 17 +- .../impl/test/EmitExecutorTest.java | 198 ++++++++++++++++++ .../emit-default-source.yaml | 13 ++ 6 files changed, 280 insertions(+), 6 deletions(-) create mode 100644 impl/core/src/main/java/io/serverlessworkflow/impl/events/EmitSourceResolver.java create mode 100644 impl/test/src/test/java/io/serverlessworkflow/impl/test/EmitExecutorTest.java create mode 100644 impl/test/src/test/resources/workflows-samples/emit-default-source.yaml diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowApplication.java b/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowApplication.java index b4b86e2bc..3d369532f 100644 --- a/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowApplication.java +++ b/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 templateResolver; private final Optional functionReader; private final URI defaultCatalogURI; + private final WorkflowValueResolver defaultEventSource; private final Collection 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 templateResolver; private Optional functionReader; private URI defaultCatalogURI; + private WorkflowValueResolver 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 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 defaultEventSource() { + return defaultEventSource; + } + public String id() { return id; } diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/events/CloudEventUtils.java b/impl/core/src/main/java/io/serverlessworkflow/impl/events/CloudEventUtils.java index 2c1f46d0e..34641513b 100644 --- a/impl/core/src/main/java/io/serverlessworkflow/impl/events/CloudEventUtils.java +++ b/impl/core/src/main/java/io/serverlessworkflow/impl/events/CloudEventUtils.java @@ -45,6 +45,6 @@ public static Map extensions(CloudEvent event) { } public static URI source() { - return URI.create("reference-impl"); + return URI.create("io.serverlessworkflow.sdk-java"); } } diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/events/EmitSourceResolver.java b/impl/core/src/main/java/io/serverlessworkflow/impl/events/EmitSourceResolver.java new file mode 100644 index 000000000..33e5b2559 --- /dev/null +++ b/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 { + + 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()); + } +} diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/EmitExecutor.java b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/EmitExecutor.java index e90d248ed..483754847 100644 --- a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/EmitExecutor.java +++ b/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())) diff --git a/impl/test/src/test/java/io/serverlessworkflow/impl/test/EmitExecutorTest.java b/impl/test/src/test/java/io/serverlessworkflow/impl/test/EmitExecutorTest.java new file mode 100644 index 000000000..3e84f18d2 --- /dev/null +++ b/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 runWorkflow(Workflow wf, String eventType) { + List 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 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 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 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 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 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 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 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 events = runWorkflow(wf, "org.example.event"); + + assertThat(events).hasSize(1); + CloudEvent ev = events.get(0); + assertThat(ev.getId()).isNotNull().isNotEmpty(); + } +} diff --git a/impl/test/src/test/resources/workflows-samples/emit-default-source.yaml b/impl/test/src/test/resources/workflows-samples/emit-default-source.yaml new file mode 100644 index 000000000..876f2905e --- /dev/null +++ b/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