Skip to content

Commit a598dab

Browse files
matheusandre1matheusandre1
committed
Add defaultEventSource to WorkflowApplication
Signed-off-by: Matheus André <matheusandr2@gmail.com>
1 parent 2417ff9 commit a598dab

6 files changed

Lines changed: 275 additions & 6 deletions

File tree

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

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@
3030
import io.serverlessworkflow.impl.config.SystemPropertyConfigManager;
3131
import io.serverlessworkflow.impl.events.CloudEventPredicateFactory;
3232
import io.serverlessworkflow.impl.events.DefaultCloudEventPredicateFactory;
33+
import io.serverlessworkflow.impl.events.EmitSourceResolver;
3334
import io.serverlessworkflow.impl.events.EventConsumer;
3435
import io.serverlessworkflow.impl.events.EventPublisher;
3536
import io.serverlessworkflow.impl.events.InMemoryEvents;
@@ -105,6 +106,7 @@ public class WorkflowApplication implements AutoCloseable {
105106
private final Optional<URITemplateResolver> templateResolver;
106107
private final Optional<FunctionReader> functionReader;
107108
private final URI defaultCatalogURI;
109+
private final WorkflowValueResolver<URI> defaultEventSource;
108110
private final Collection<CallableTaskProxyBuilder> callableProxyBuilders;
109111
private final CloudEventPredicateFactory cloudEventPredicateFactory;
110112
private final AllStrategyCorrelationInfoFactory allStrategyCorrelationInfoFactory;
@@ -138,6 +140,7 @@ private WorkflowApplication(Builder builder) {
138140
this.templateResolver = builder.templateResolver;
139141
this.functionReader = builder.functionReader;
140142
this.defaultCatalogURI = builder.defaultCatalogURI;
143+
this.defaultEventSource = builder.defaultEventSource;
141144
this.id = builder.id;
142145
this.callableProxyBuilders = builder.callableProxyBuilders;
143146
this.cloudEventPredicateFactory = builder.cloudEventPredicateFactory;
@@ -263,6 +266,7 @@ public SchemaValidator getValidator(SchemaInline inline) {
263266
private Optional<URITemplateResolver> templateResolver;
264267
private Optional<FunctionReader> functionReader;
265268
private URI defaultCatalogURI;
269+
private WorkflowValueResolver<URI> defaultEventSource;
266270
private CloudEventPredicateFactory cloudEventPredicateFactory;
267271
private AllStrategyCorrelationInfoFactory allStrategyCorrelationInfoFactory;
268272
private WorkflowLifeCycleCloudEventFactory lifeCycleCloudEventFactory;
@@ -404,6 +408,15 @@ public Builder withDefaultCatalogURI(URI defaultCatalogURI) {
404408
return this;
405409
}
406410

411+
public Builder withDefaultEventSource(URI defaultEventSource) {
412+
return withDefaultEventSource((workflow, task, model) -> defaultEventSource);
413+
}
414+
415+
public Builder withDefaultEventSource(WorkflowValueResolver<URI> defaultEventSource) {
416+
this.defaultEventSource = defaultEventSource;
417+
return this;
418+
}
419+
407420
public Builder withCloudEventPredicateFactory(
408421
CloudEventPredicateFactory cloudEventPredicateFactory) {
409422
this.cloudEventPredicateFactory = cloudEventPredicateFactory;
@@ -521,6 +534,9 @@ public CronResolver parseCron(String cron) {
521534
if (defaultCatalogURI == null) {
522535
defaultCatalogURI = URI.create("https://github.com/serverlessworkflow/catalog");
523536
}
537+
if (defaultEventSource == null) {
538+
defaultEventSource = new EmitSourceResolver();
539+
}
524540
Collections.sort(listeners);
525541
Collections.sort(callableProxyBuilders);
526542
if (id == null) {
@@ -639,6 +655,10 @@ public URI defaultCatalogURI() {
639655
return defaultCatalogURI;
640656
}
641657

658+
public WorkflowValueResolver<URI> defaultEventSource() {
659+
return defaultEventSource;
660+
}
661+
642662
public String id() {
643663
return id;
644664
}

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -45,6 +45,6 @@ public static Map<String, Object> extensions(CloudEvent event) {
4545
}
4646

4747
public static URI source() {
48-
return URI.create("reference-impl");
48+
return URI.create("io.serverlessworkflow.sdk-java");
4949
}
5050
}

‎impl/core/src/main/java/io/serverlessworkflow/impl/events/EmitSourceResolver.java‎

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,33 @@
1+
/*
2+
* Copyright 2020-Present The Serverless Workflow Specification Authors
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
package io.serverlessworkflow.impl.events;
17+
18+
import io.serverlessworkflow.impl.TaskContext;
19+
import io.serverlessworkflow.impl.WorkflowContext;
20+
import io.serverlessworkflow.impl.WorkflowModel;
21+
import io.serverlessworkflow.impl.WorkflowValueResolver;
22+
import java.net.URI;
23+
24+
public class EmitSourceResolver implements WorkflowValueResolver<URI> {
25+
26+
private static final String SEP = "/";
27+
28+
@Override
29+
public URI apply(WorkflowContext workflow, TaskContext task, WorkflowModel model) {
30+
return URI.create(
31+
workflow.definition().application().id() + SEP + workflow.definition().id().toString(SEP));
32+
}
33+
}

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

Lines changed: 12 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -100,17 +100,24 @@ private CloudEvent buildCloudEvent(WorkflowContext workflow, TaskContext taskCon
100100
.sourceFilter()
101101
.map(filter -> filter.apply(workflow, taskContext, taskContext.input()))
102102
.map(URI::create)
103-
.orElse(CloudEventUtils.source()));
103+
.orElseGet(
104+
() ->
105+
workflow
106+
.definition()
107+
.application()
108+
.defaultEventSource()
109+
.apply(workflow, taskContext, taskContext.input())));
104110
ceBuilder.withType(
105111
props
106112
.typeFilter()
107113
.map(filter -> filter.apply(workflow, taskContext, taskContext.input()))
108114
.orElseThrow(
109115
() -> new IllegalArgumentException("Type is required for emitting events")));
110-
props
111-
.timeFilter()
112-
.map(filter -> filter.apply(workflow, taskContext, taskContext.input()))
113-
.ifPresent(value -> ceBuilder.withTime(value));
116+
ceBuilder.withTime(
117+
props
118+
.timeFilter()
119+
.map(filter -> filter.apply(workflow, taskContext, taskContext.input()))
120+
.orElseGet(OffsetDateTime::now));
114121
props
115122
.subjectFilter()
116123
.map(filter -> filter.apply(workflow, taskContext, taskContext.input()))

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

Lines changed: 196 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,196 @@
1+
/*
2+
* Copyright 2020-Present The Serverless Workflow Specification Authors
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
package io.serverlessworkflow.impl.test;
17+
18+
import static io.serverlessworkflow.api.WorkflowReader.readWorkflowFromClasspath;
19+
import static org.assertj.core.api.Assertions.assertThat;
20+
import static org.assertj.core.api.Assertions.assertThatThrownBy;
21+
22+
import io.cloudevents.CloudEvent;
23+
import io.serverlessworkflow.api.types.Workflow;
24+
import io.serverlessworkflow.fluent.spec.EventPropertiesBuilder;
25+
import io.serverlessworkflow.fluent.spec.WorkflowBuilder;
26+
import io.serverlessworkflow.impl.WorkflowApplication;
27+
import io.serverlessworkflow.impl.events.InMemoryEvents;
28+
import java.io.IOException;
29+
import java.net.URI;
30+
import java.time.OffsetDateTime;
31+
import java.util.ArrayList;
32+
import java.util.Collections;
33+
import java.util.List;
34+
import java.util.function.Consumer;
35+
import java.util.function.UnaryOperator;
36+
import org.junit.jupiter.api.BeforeEach;
37+
import org.junit.jupiter.api.Test;
38+
39+
class EmitExecutorTest {
40+
41+
private static final String APP_ID = "test-app";
42+
43+
private List<CloudEvent> events;
44+
private InMemoryEvents broker;
45+
46+
@BeforeEach
47+
void setUp() {
48+
events = Collections.synchronizedList(new ArrayList<>());
49+
broker = new InMemoryEvents();
50+
}
51+
52+
private List<CloudEvent> runWorkflow(Workflow wf, String eventType) {
53+
return runWorkflow(wf, eventType, UnaryOperator.identity());
54+
}
55+
56+
private List<CloudEvent> runWorkflow(
57+
Workflow wf, String eventType, UnaryOperator<WorkflowApplication.Builder> customizer) {
58+
broker.register(eventType, events::add);
59+
try (WorkflowApplication app =
60+
customizer
61+
.apply(
62+
WorkflowApplication.builder()
63+
.withId(APP_ID)
64+
.withEventConsumer(broker)
65+
.withEventPublisher(broker))
66+
.build()) {
67+
app.workflowDefinition(wf).instance().start().join();
68+
}
69+
return events;
70+
}
71+
72+
private static Workflow emitWorkflow(String name, Consumer<EventPropertiesBuilder> props) {
73+
return WorkflowBuilder.workflow(name, "test", "0.1.0")
74+
.tasks(t -> t.emit("emitEvent", etb -> etb.event(props::accept)))
75+
.build();
76+
}
77+
78+
private static Workflow emitWorkflow(String name, String type) {
79+
return emitWorkflow(name, epb -> epb.type(type));
80+
}
81+
82+
@Test
83+
void whenMissingType_throwsIllegalArgumentException() throws IOException {
84+
Workflow wf =
85+
WorkflowBuilder.workflow("emit-no-type", "test", "0.1.0")
86+
.tasks(t -> t.emit("emitEvent", etb -> etb.event(epb -> {})))
87+
.build();
88+
89+
assertThatThrownBy(() -> runWorkflow(wf, "any"))
90+
.isInstanceOf(java.util.concurrent.CompletionException.class)
91+
.hasCauseInstanceOf(IllegalArgumentException.class)
92+
.hasMessageContaining("Type is required for emitting events");
93+
}
94+
95+
@Test
96+
void whenTypeOnly_sourceCombinesApplicationIdAndDefinitionId() throws IOException {
97+
Workflow wf = emitWorkflow("emit-source-resolver", "org.example.event");
98+
List<CloudEvent> events = runWorkflow(wf, "org.example.event");
99+
100+
assertThat(events).hasSize(1);
101+
CloudEvent ev = events.get(0);
102+
assertThat(ev.getType()).isEqualTo("org.example.event");
103+
assertThat(ev.getSource()).isEqualTo(URI.create(APP_ID + "/test/emit-source-resolver/0.1.0"));
104+
}
105+
106+
@Test
107+
void whenAppHasDefaultEventSource_itOverridesCalculatedDefault() throws IOException {
108+
Workflow wf = emitWorkflow("emit-app-default", "org.example.event");
109+
List<CloudEvent> events =
110+
runWorkflow(
111+
wf, "org.example.event", b -> b.withDefaultEventSource(URI.create("my.custom.source")));
112+
113+
assertThat(events).hasSize(1);
114+
CloudEvent ev = events.get(0);
115+
assertThat(ev.getSource()).isEqualTo(URI.create("my.custom.source"));
116+
}
117+
118+
@Test
119+
void whenExplicitSourceInEventDefinition_itTakesPrecedenceOverAllDefaults() throws IOException {
120+
Workflow wf =
121+
WorkflowBuilder.workflow("emit-explicit-source", "my.ns", "1.2.3")
122+
.tasks(
123+
t ->
124+
t.emit(
125+
"emitEvent",
126+
etb ->
127+
etb.event(
128+
epb -> {
129+
epb.type("org.example.event");
130+
epb.source("explicit/source");
131+
})))
132+
.build();
133+
134+
List<CloudEvent> events =
135+
runWorkflow(
136+
wf, "org.example.event", b -> b.withDefaultEventSource(URI.create("app.source")));
137+
138+
assertThat(events).hasSize(1);
139+
CloudEvent ev = events.get(0);
140+
assertThat(ev.getSource()).isEqualTo(URI.create("explicit/source"));
141+
}
142+
143+
@Test
144+
void whenDefaultEventSourceIsResolver_uriIsResolvedFromWorkflowContext() throws IOException {
145+
Workflow wf = emitWorkflow("emit-resolver-default", "org.example.event");
146+
List<CloudEvent> events =
147+
runWorkflow(
148+
wf,
149+
"org.example.event",
150+
b ->
151+
b.withDefaultEventSource(
152+
(workflow, task, model) ->
153+
URI.create("apps/" + workflow.definition().id().name())));
154+
155+
assertThat(events).hasSize(1);
156+
CloudEvent ev = events.get(0);
157+
assertThat(ev.getSource()).isEqualTo(URI.create("apps/emit-resolver-default"));
158+
}
159+
160+
@Test
161+
void whenWorkflowIsLoadedFromYaml_sourceCombinesApplicationIdAndDefinitionId()
162+
throws IOException {
163+
Workflow wf = readWorkflowFromClasspath("workflows-samples/emit-default-source.yaml");
164+
List<CloudEvent> events = runWorkflow(wf, "com.test.default.source");
165+
166+
assertThat(events).hasSize(1);
167+
CloudEvent ev = events.get(0);
168+
assertThat(ev.getType()).isEqualTo("com.test.default.source");
169+
assertThat(ev.getSource()).isEqualTo(URI.create(APP_ID + "/test/emit-default-source/0.1.0"));
170+
assertThat(ev.getTime()).isNotNull();
171+
assertThat(ev.getId()).isNotNull().isNotEmpty();
172+
}
173+
174+
@Test
175+
void whenTimeIsNotSpecified_defaultsToNow() throws IOException {
176+
Workflow wf = emitWorkflow("emit-time-default", "org.example.event");
177+
List<CloudEvent> events = runWorkflow(wf, "org.example.event");
178+
179+
assertThat(events).hasSize(1);
180+
CloudEvent ev = events.get(0);
181+
OffsetDateTime now = OffsetDateTime.now();
182+
assertThat(ev.getTime()).isNotNull();
183+
assertThat(ev.getTime()).isBeforeOrEqualTo(now);
184+
assertThat(ev.getTime()).isAfter(now.minusSeconds(30));
185+
}
186+
187+
@Test
188+
void whenIdIsNotSpecified_autoGeneratedIdIsSet() throws IOException {
189+
Workflow wf = emitWorkflow("emit-id-default", "org.example.event");
190+
List<CloudEvent> events = runWorkflow(wf, "org.example.event");
191+
192+
assertThat(events).hasSize(1);
193+
CloudEvent ev = events.get(0);
194+
assertThat(ev.getId()).isNotNull().isNotEmpty();
195+
}
196+
}

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

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,13 @@
1+
document:
2+
dsl: '1.0.0-alpha5'
3+
namespace: test
4+
name: emit-default-source
5+
version: '0.1.0'
6+
do:
7+
- emitEvent:
8+
emit:
9+
event:
10+
with:
11+
type: com.test.default.source
12+
data:
13+
message: hello

0 commit comments

Comments
 (0)