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..54dd4186d 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; private CloudEventPredicateFactory cloudEventPredicateFactory; private AllStrategyCorrelationInfoFactory allStrategyCorrelationInfoFactory; private WorkflowLifeCycleCloudEventFactory lifeCycleCloudEventFactory; @@ -404,6 +408,15 @@ public Builder withDefaultCatalogURI(URI defaultCatalogURI) { return this; } + 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; @@ -521,6 +534,9 @@ public CronResolver parseCron(String cron) { if (defaultCatalogURI == null) { defaultCatalogURI = URI.create("https://github.com/serverlessworkflow/catalog"); } + if (defaultEventSource == null) { + defaultEventSource = new EmitSourceResolver(); + } Collections.sort(listeners); Collections.sort(callableProxyBuilders); if (id == null) { @@ -639,6 +655,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..6ea9ee232 --- /dev/null +++ b/impl/core/src/main/java/io/serverlessworkflow/impl/events/EmitSourceResolver.java @@ -0,0 +1,33 @@ +/* + * 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.WorkflowModel; +import io.serverlessworkflow.impl.WorkflowValueResolver; +import java.net.URI; + +public class EmitSourceResolver implements WorkflowValueResolver { + + private static final String SEP = "/"; + + @Override + public URI apply(WorkflowContext workflow, TaskContext task, WorkflowModel model) { + return URI.create( + workflow.definition().application().id() + SEP + workflow.definition().id().toString(SEP)); + } +} 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..8615c66d0 --- /dev/null +++ b/impl/test/src/test/java/io/serverlessworkflow/impl/test/EmitExecutorTest.java @@ -0,0 +1,196 @@ +/* + * 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 java.util.function.UnaryOperator; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +class EmitExecutorTest { + + private static final String APP_ID = "test-app"; + + private List events; + private InMemoryEvents broker; + + @BeforeEach + void setUp() { + events = Collections.synchronizedList(new ArrayList<>()); + broker = new InMemoryEvents(); + } + + private List runWorkflow(Workflow wf, String eventType) { + return runWorkflow(wf, eventType, UnaryOperator.identity()); + } + + private List runWorkflow( + Workflow wf, String eventType, UnaryOperator customizer) { + broker.register(eventType, events::add); + try (WorkflowApplication app = + customizer + .apply( + WorkflowApplication.builder() + .withId(APP_ID) + .withEventConsumer(broker) + .withEventPublisher(broker)) + .build()) { + app.workflowDefinition(wf).instance().start().join(); + } + 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_sourceCombinesApplicationIdAndDefinitionId() 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(APP_ID + "/test/emit-source-resolver/0.1.0")); + } + + @Test + void whenAppHasDefaultEventSource_itOverridesCalculatedDefault() throws IOException { + Workflow wf = emitWorkflow("emit-app-default", "org.example.event"); + List events = + runWorkflow( + wf, "org.example.event", b -> b.withDefaultEventSource(URI.create("my.custom.source"))); + + 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 = + runWorkflow( + wf, "org.example.event", b -> b.withDefaultEventSource(URI.create("app.source"))); + + 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 = + runWorkflow( + wf, + "org.example.event", + b -> + b.withDefaultEventSource( + (workflow, task, model) -> + URI.create("apps/" + workflow.definition().id().name()))); + + assertThat(events).hasSize(1); + CloudEvent ev = events.get(0); + assertThat(ev.getSource()).isEqualTo(URI.create("apps/emit-resolver-default")); + } + + @Test + void whenWorkflowIsLoadedFromYaml_sourceCombinesApplicationIdAndDefinitionId() + 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(APP_ID + "/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