Skip to content

Commit 856013f

Browse files
mcruzdevmcruzdev
committed
Add initial asyncapi implementation
Signed-off-by: Matheus Cruz <matheuscruz.dev@gmail.com>
1 parent afddf80 commit 856013f

19 files changed

Lines changed: 1587 additions & 4 deletions

‎asyncapi-call-plan.md‎

Lines changed: 488 additions & 0 deletions
Large diffs are not rendered by default.

‎impl/asyncapi/pom.xml‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -14,5 +14,10 @@
1414
<artifactId>serverlessworkflow-impl-core</artifactId>
1515
<version>${project.version}</version>
1616
</dependency>
17+
<dependency>
18+
<groupId>io.serverlessworkflow</groupId>
19+
<artifactId>serverlessworkflow-api</artifactId>
20+
<version>${project.version}</version>
21+
</dependency>
1722
</dependencies>
1823
</project>

‎impl/asyncapi/src/main/java/io/serverlessworkflow/impl/executors/asyncapi/AsyncAPIExecutor.java‎

Lines changed: 300 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -15,17 +15,315 @@
1515
*/
1616
package io.serverlessworkflow.impl.executors.asyncapi;
1717

18+
import io.serverlessworkflow.api.types.AsyncApiArguments.AsyncApiProtocol;
19+
import io.serverlessworkflow.api.types.AsyncApiMessageConsumptionPolicyAmount;
20+
import io.serverlessworkflow.api.types.AsyncApiMessageConsumptionPolicyUnion;
21+
import io.serverlessworkflow.api.types.AsyncApiServer;
22+
import io.serverlessworkflow.api.types.ExternalResource;
1823
import io.serverlessworkflow.impl.TaskContext;
1924
import io.serverlessworkflow.impl.WorkflowContext;
2025
import io.serverlessworkflow.impl.WorkflowModel;
26+
import io.serverlessworkflow.impl.WorkflowModelCollection;
27+
import io.serverlessworkflow.impl.WorkflowModelFactory;
28+
import io.serverlessworkflow.impl.WorkflowPredicate;
29+
import io.serverlessworkflow.impl.WorkflowValueResolver;
30+
import io.serverlessworkflow.impl.auth.AuthProvider;
2131
import io.serverlessworkflow.impl.executors.CallableTask;
32+
import io.serverlessworkflow.impl.executors.TaskExecutor;
33+
import io.serverlessworkflow.impl.executors.TaskExecutorHelper;
34+
import java.net.URI;
35+
import java.time.Duration;
36+
import java.util.Collections;
37+
import java.util.HashMap;
38+
import java.util.Map;
39+
import java.util.Optional;
2240
import java.util.concurrent.CompletableFuture;
41+
import java.util.concurrent.TimeUnit;
2342

24-
public class AsyncAPIExecutor implements CallableTask {
43+
class AsyncAPIExecutor implements CallableTask {
44+
45+
record PublishConfig(
46+
WorkflowValueResolver<Map<String, Object>> payloadResolver,
47+
WorkflowValueResolver<Map<String, Object>> headersResolver) {}
48+
49+
record SubscribeConfig(
50+
Optional<WorkflowPredicate> filterPredicate,
51+
AsyncApiMessageConsumptionPolicyUnion consumePolicy,
52+
Optional<WorkflowValueResolver<Duration>> consumeTimeout,
53+
Optional<WorkflowPredicate> whilePredicate,
54+
Optional<WorkflowPredicate> untilPredicate,
55+
TaskExecutor<?> foreachExecutor,
56+
String foreachItem,
57+
String foreachAt) {}
58+
59+
private final ExternalResource document;
60+
private final String operationName;
61+
private final String channelName;
62+
private final AsyncApiServer serverConfig;
63+
private final AsyncApiProtocol protocolConfig;
64+
private final Optional<AuthProvider> authProvider;
65+
private final PublishConfig publishConfig;
66+
private final SubscribeConfig subscribeConfig;
67+
68+
AsyncAPIExecutor(
69+
ExternalResource document,
70+
String operationName,
71+
String channelName,
72+
AsyncApiServer serverConfig,
73+
AsyncApiProtocol protocolConfig,
74+
Optional<AuthProvider> authProvider,
75+
PublishConfig publishConfig,
76+
SubscribeConfig subscribeConfig) {
77+
this.document = document;
78+
this.operationName = operationName;
79+
this.channelName = channelName;
80+
this.serverConfig = serverConfig;
81+
this.protocolConfig = protocolConfig;
82+
this.authProvider = authProvider;
83+
this.publishConfig = publishConfig;
84+
this.subscribeConfig = subscribeConfig;
85+
}
2586

2687
@Override
2788
public CompletableFuture<WorkflowModel> apply(
2889
WorkflowContext workflowContext, TaskContext taskContext, WorkflowModel input) {
29-
return CompletableFuture.completedFuture(input);
90+
return CompletableFuture.supplyAsync(
91+
() -> loadAndResolve(workflowContext, taskContext, input),
92+
workflowContext.definition().application().executorService())
93+
.thenCompose(
94+
channelInfo -> {
95+
AsyncApiChannelProvider provider = lookupProvider(workflowContext, taskContext);
96+
if (publishConfig != null) {
97+
return doPublish(provider, channelInfo, workflowContext, taskContext, input);
98+
} else {
99+
return doSubscribe(provider, channelInfo, workflowContext, taskContext, input);
100+
}
101+
});
102+
}
103+
104+
private AsyncApiChannelInfo loadAndResolve(
105+
WorkflowContext workflowContext, TaskContext taskContext, WorkflowModel input) {
106+
UnifiedAsyncAPI asyncApi =
107+
workflowContext
108+
.definition()
109+
.resourceLoader()
110+
.load(document, AsyncAPIReader::read, workflowContext, taskContext, input);
111+
112+
UnifiedAsyncAPI.Server server = resolveServer(asyncApi);
113+
String resolvedChannel = resolveChannel(asyncApi);
114+
String url = substituteVariables(server.effectiveUrl(), server);
115+
URI serverUri = URI.create(server.protocol() + "://" + url);
116+
117+
Optional<String> authToken =
118+
authProvider.map(
119+
auth -> auth.content(workflowContext, taskContext, input, serverUri).join());
120+
121+
return new AsyncApiChannelInfo(
122+
serverUri,
123+
resolvedChannel,
124+
operationName != null ? operationName : channelName,
125+
server.protocol(),
126+
authToken);
127+
}
128+
129+
private UnifiedAsyncAPI.Server resolveServer(UnifiedAsyncAPI asyncApi) {
130+
if (asyncApi.servers() == null || asyncApi.servers().isEmpty()) {
131+
throw new IllegalArgumentException("AsyncAPI document has no servers defined");
132+
}
133+
if (serverConfig != null && serverConfig.getName() != null) {
134+
UnifiedAsyncAPI.Server server = asyncApi.servers().get(serverConfig.getName());
135+
if (server != null) {
136+
return server;
137+
}
138+
throw new IllegalArgumentException(
139+
"Server '" + serverConfig.getName() + "' not found in AsyncAPI document");
140+
}
141+
if (protocolConfig != null) {
142+
String proto = protocolConfig.value();
143+
return asyncApi.servers().values().stream()
144+
.filter(s -> proto.equals(s.protocol()))
145+
.findFirst()
146+
.orElseThrow(
147+
() ->
148+
new IllegalArgumentException(
149+
"No server with protocol '" + proto + "' in AsyncAPI document"));
150+
}
151+
return asyncApi.servers().values().iterator().next();
152+
}
153+
154+
private String resolveChannel(UnifiedAsyncAPI asyncApi) {
155+
if (channelName != null) {
156+
return channelName;
157+
}
158+
if (operationName != null && asyncApi.isV3() && asyncApi.operations() != null) {
159+
UnifiedAsyncAPI.Operation op = asyncApi.operations().get(operationName);
160+
if (op != null && op.channel() != null) {
161+
String name = op.channel().channelName();
162+
if (asyncApi.channels() != null && asyncApi.channels().containsKey(name)) {
163+
UnifiedAsyncAPI.Channel ch = asyncApi.channels().get(name);
164+
return ch.address() != null ? ch.address() : name;
165+
}
166+
return name;
167+
}
168+
}
169+
throw new IllegalArgumentException(
170+
"Cannot resolve channel: provide 'channel' (v2) or 'operation' (v3)");
171+
}
172+
173+
private String substituteVariables(String url, UnifiedAsyncAPI.Server docServer) {
174+
if (serverConfig != null
175+
&& serverConfig.getVariables() != null
176+
&& serverConfig.getVariables().getAdditionalProperties() != null) {
177+
for (Map.Entry<String, Object> entry :
178+
serverConfig.getVariables().getAdditionalProperties().entrySet()) {
179+
url = url.replace("{" + entry.getKey() + "}", entry.getValue().toString());
180+
}
181+
}
182+
if (docServer.variables() != null) {
183+
for (Map.Entry<String, UnifiedAsyncAPI.ServerVariable> entry :
184+
docServer.variables().entrySet()) {
185+
if (entry.getValue().defaultValue() != null) {
186+
url = url.replace("{" + entry.getKey() + "}", entry.getValue().defaultValue());
187+
}
188+
}
189+
}
190+
return url;
191+
}
192+
193+
private CompletableFuture<WorkflowModel> doPublish(
194+
AsyncApiChannelProvider provider,
195+
AsyncApiChannelInfo channelInfo,
196+
WorkflowContext workflowContext,
197+
TaskContext taskContext,
198+
WorkflowModel input) {
199+
Map<String, Object> payload =
200+
publishConfig.payloadResolver() != null
201+
? publishConfig.payloadResolver().apply(workflowContext, taskContext, input)
202+
: Collections.emptyMap();
203+
Map<String, Object> headers =
204+
publishConfig.headersResolver() != null
205+
? publishConfig.headersResolver().apply(workflowContext, taskContext, input)
206+
: Collections.emptyMap();
207+
return provider.publish(channelInfo, payload, headers).thenApply(v -> input);
208+
}
209+
210+
private CompletableFuture<WorkflowModel> doSubscribe(
211+
AsyncApiChannelProvider provider,
212+
AsyncApiChannelInfo channelInfo,
213+
WorkflowContext workflowContext,
214+
TaskContext taskContext,
215+
WorkflowModel input) {
216+
WorkflowModelFactory factory = workflowContext.definition().application().modelFactory();
217+
WorkflowModelCollection collection = factory.createCollection();
218+
CompletableFuture<WorkflowModel> result = new CompletableFuture<>();
219+
220+
AsyncApiSubscriptionHandle handle =
221+
provider.subscribe(
222+
channelInfo,
223+
msg -> {
224+
synchronized (collection) {
225+
if (result.isDone()) {
226+
return;
227+
}
228+
WorkflowModel messageModel = toWorkflowModel(factory, msg);
229+
if (subscribeConfig.filterPredicate().isPresent()
230+
&& !subscribeConfig
231+
.filterPredicate()
232+
.get()
233+
.test(workflowContext, taskContext, messageModel)) {
234+
return;
235+
}
236+
WorkflowModel processedModel =
237+
processMessage(messageModel, collection, workflowContext, taskContext);
238+
collection.add(processedModel);
239+
if (isConsumptionPolicySatisfied(workflowContext, taskContext, collection)) {
240+
result.complete(collection);
241+
}
242+
}
243+
});
244+
245+
result.whenComplete((r, ex) -> handle.unsubscribe());
246+
247+
subscribeConfig
248+
.consumeTimeout()
249+
.ifPresent(
250+
resolver -> {
251+
Duration duration = resolver.apply(workflowContext, taskContext, input);
252+
CompletableFuture.delayedExecutor(duration.toMillis(), TimeUnit.MILLISECONDS)
253+
.execute(
254+
() -> {
255+
synchronized (collection) {
256+
if (!result.isDone()) {
257+
result.complete(collection);
258+
}
259+
}
260+
});
261+
});
262+
263+
return result;
264+
}
265+
266+
private WorkflowModel processMessage(
267+
WorkflowModel messageModel,
268+
WorkflowModelCollection collection,
269+
WorkflowContext workflowContext,
270+
TaskContext taskContext) {
271+
if (subscribeConfig.foreachExecutor() != null) {
272+
taskContext.variables().put(subscribeConfig.foreachItem(), messageModel);
273+
taskContext.variables().put(subscribeConfig.foreachAt(), collection.size());
274+
return TaskExecutorHelper.processTaskList(
275+
subscribeConfig.foreachExecutor(),
276+
workflowContext,
277+
Optional.of(taskContext),
278+
messageModel)
279+
.join();
280+
}
281+
return messageModel;
282+
}
283+
284+
private WorkflowModel toWorkflowModel(WorkflowModelFactory factory, AsyncApiInboundMessage msg) {
285+
Map<String, Object> map = new HashMap<>();
286+
map.put("payload", msg.payload());
287+
map.put("headers", msg.headers());
288+
msg.correlationId().ifPresent(id -> map.put("correlationId", id));
289+
return factory.from(map);
290+
}
291+
292+
private boolean isConsumptionPolicySatisfied(
293+
WorkflowContext workflowContext,
294+
TaskContext taskContext,
295+
WorkflowModelCollection collection) {
296+
AsyncApiMessageConsumptionPolicyUnion policy = subscribeConfig.consumePolicy();
297+
if (policy == null) {
298+
return false;
299+
}
300+
AsyncApiMessageConsumptionPolicyAmount amount =
301+
policy.getAsyncApiMessageConsumptionPolicyAmount();
302+
if (amount != null) {
303+
return collection.size() >= amount.getAmount();
304+
}
305+
if (subscribeConfig.whilePredicate().isPresent()) {
306+
return !subscribeConfig.whilePredicate().get().test(workflowContext, taskContext, collection);
307+
}
308+
if (subscribeConfig.untilPredicate().isPresent()) {
309+
return subscribeConfig.untilPredicate().get().test(workflowContext, taskContext, collection);
310+
}
311+
return false;
312+
}
313+
314+
private AsyncApiChannelProvider lookupProvider(
315+
WorkflowContext workflowContext, TaskContext taskContext) {
316+
return workflowContext
317+
.definition()
318+
.application()
319+
.<AsyncApiChannelProvider>additionalObject(
320+
AsyncApiChannelProvider.ASYNC_API_CHANNEL_PROVIDER, workflowContext, taskContext)
321+
.orElseThrow(
322+
() ->
323+
new IllegalStateException(
324+
"Missing AsyncApiChannelProvider. Register one via"
325+
+ " WorkflowApplication.builder().withAdditionalObject(\""
326+
+ AsyncApiChannelProvider.ASYNC_API_CHANNEL_PROVIDER
327+
+ "\", provider)"));
30328
}
31329
}

0 commit comments

Comments
 (0)