diff --git a/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/BaseTaskItemListBuilder.java b/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/BaseTaskItemListBuilder.java index 76a0dc57d..28f7ec072 100644 --- a/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/BaseTaskItemListBuilder.java +++ b/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/BaseTaskItemListBuilder.java @@ -46,6 +46,7 @@ public abstract class BaseTaskItemListBuilder list; diff --git a/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/BaseWorkflowBuilder.java b/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/BaseWorkflowBuilder.java index 329e486e7..365322618 100644 --- a/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/BaseWorkflowBuilder.java +++ b/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/BaseWorkflowBuilder.java @@ -136,7 +136,11 @@ private SELF appendDo(Consumer configurer) { configurer.accept(doBuilder); final List newItems = doBuilder.build().getDo(); - if (newItems == null || newItems.isEmpty()) return self(); + if (newItems == null || newItems.isEmpty()) { + throw new IllegalStateException( + "Task list must contain at least one task. " + + "Use .tasks(d -> d.set(...)) or similar to define tasks."); + } final List merged = new ArrayList<>(this.workflow.getDo() != null ? this.workflow.getDo() : List.of()); diff --git a/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/CallAsyncAPITaskBuilder.java b/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/CallAsyncAPITaskBuilder.java new file mode 100644 index 000000000..0ff475bed --- /dev/null +++ b/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/CallAsyncAPITaskBuilder.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.fluent.spec; + +import io.serverlessworkflow.api.types.AsyncApiArguments; +import io.serverlessworkflow.api.types.CallAsyncAPI; +import io.serverlessworkflow.fluent.spec.spi.CallAsyncAPITaskFluent; + +public class CallAsyncAPITaskBuilder extends TaskBaseBuilder + implements CallAsyncAPITaskFluent { + + CallAsyncAPITaskBuilder() { + final CallAsyncAPI callAsyncAPI = new CallAsyncAPI(); + callAsyncAPI.setWith(new AsyncApiArguments()); + super.setTask(callAsyncAPI); + } + + @Override + public CallAsyncAPITaskBuilder self() { + return this; + } +} diff --git a/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/DoTaskBuilder.java b/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/DoTaskBuilder.java index 4199811e3..d28279477 100644 --- a/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/DoTaskBuilder.java +++ b/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/DoTaskBuilder.java @@ -104,6 +104,12 @@ public DoTaskBuilder openapi(String name, Consumer items return this; } + @Override + public DoTaskBuilder asyncapi(String name, Consumer itemsConfigurer) { + this.listBuilder().asyncapi(name, itemsConfigurer); + return this; + } + @Override public DoTaskBuilder grpc(String name, Consumer itemsConfigurer) { this.listBuilder().grpc(name, itemsConfigurer); diff --git a/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/TaskItemListBuilder.java b/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/TaskItemListBuilder.java index d227718f3..a1a962d58 100644 --- a/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/TaskItemListBuilder.java +++ b/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/TaskItemListBuilder.java @@ -154,6 +154,22 @@ public TaskItemListBuilder openapi( return addTaskItem(new TaskItem(name, task)); } + @Override + public TaskItemListBuilder asyncapi( + String name, Consumer itemsConfigurer) { + name = defaultNameAndRequireConfig(name, itemsConfigurer, TYPE_ASYNCAPI); + + final CallAsyncAPITaskBuilder callAsyncAPIBuilder = new CallAsyncAPITaskBuilder(); + itemsConfigurer.accept(callAsyncAPIBuilder); + + final CallTask callTask = new CallTask(); + callTask.setCallAsyncAPI(callAsyncAPIBuilder.build()); + final Task task = new Task(); + task.setCallTask(callTask); + + return addTaskItem(new TaskItem(name, task)); + } + @Override public TaskItemListBuilder grpc(String name, Consumer itemsConfigurer) { name = defaultNameAndRequireConfig(name, itemsConfigurer, TYPE_GRPC); diff --git a/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/configurers/CallAsyncAPIConfigurer.java b/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/configurers/CallAsyncAPIConfigurer.java new file mode 100644 index 000000000..adbbe78bc --- /dev/null +++ b/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/configurers/CallAsyncAPIConfigurer.java @@ -0,0 +1,22 @@ +/* + * 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.fluent.spec.configurers; + +import io.serverlessworkflow.fluent.spec.CallAsyncAPITaskBuilder; +import java.util.function.Consumer; + +@FunctionalInterface +public interface CallAsyncAPIConfigurer extends Consumer {} diff --git a/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/dsl/CallAsyncAPISpec.java b/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/dsl/CallAsyncAPISpec.java new file mode 100644 index 000000000..029982777 --- /dev/null +++ b/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/dsl/CallAsyncAPISpec.java @@ -0,0 +1,137 @@ +/* + * 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.fluent.spec.dsl; + +import io.serverlessworkflow.api.types.AsyncApiArguments; +import io.serverlessworkflow.fluent.spec.CallAsyncAPITaskBuilder; +import io.serverlessworkflow.fluent.spec.SubscriptionIteratorBuilder; +import io.serverlessworkflow.fluent.spec.TaskItemListBuilder; +import io.serverlessworkflow.fluent.spec.configurers.AuthenticationConfigurer; +import io.serverlessworkflow.fluent.spec.configurers.CallAsyncAPIConfigurer; +import io.serverlessworkflow.fluent.spec.spi.CallAsyncAPITaskFluent; +import java.net.URI; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.function.Consumer; + +public final class CallAsyncAPISpec implements CallAsyncAPIConfigurer { + + private final List>> steps = new ArrayList<>(); + + public CallAsyncAPISpec document(String uri) { + steps.add(b -> b.document(uri)); + return this; + } + + public CallAsyncAPISpec document(String uri, AuthenticationConfigurer authenticationConfigurer) { + steps.add(b -> b.document(uri, authenticationConfigurer)); + return this; + } + + public CallAsyncAPISpec document(URI uri) { + steps.add(b -> b.document(uri)); + return this; + } + + public CallAsyncAPISpec document(URI uri, AuthenticationConfigurer authenticationConfigurer) { + steps.add(b -> b.document(uri, authenticationConfigurer)); + return this; + } + + public CallAsyncAPISpec channel(String channel) { + steps.add(b -> b.channel(channel)); + return this; + } + + public CallAsyncAPISpec operation(String operation) { + steps.add(b -> b.operation(operation)); + return this; + } + + public CallAsyncAPISpec server(String name) { + steps.add(b -> b.server(name)); + return this; + } + + public CallAsyncAPISpec server(String name, Map variables) { + steps.add(b -> b.server(name, variables)); + return this; + } + + public CallAsyncAPISpec protocol(AsyncApiArguments.AsyncApiProtocol protocol) { + steps.add(b -> b.protocol(protocol)); + return this; + } + + public CallAsyncAPISpec message(Map payload) { + steps.add(b -> b.message(payload)); + return this; + } + + public CallAsyncAPISpec message(Map payload, Map headers) { + steps.add(b -> b.message(payload, headers)); + return this; + } + + public CallAsyncAPISpec payload(Map payload) { + steps.add(b -> b.payload(payload)); + return this; + } + + public CallAsyncAPISpec headers(Map headers) { + steps.add(b -> b.headers(headers)); + return this; + } + + public CallAsyncAPISpec consumeAmount(int amount) { + steps.add(b -> b.consumeAmount(amount)); + return this; + } + + public CallAsyncAPISpec consumeWhile(String expression) { + steps.add(b -> b.consumeWhile(expression)); + return this; + } + + public CallAsyncAPISpec consumeUntil(String expression) { + steps.add(b -> b.consumeUntil(expression)); + return this; + } + + public CallAsyncAPISpec filter(String filterExpression) { + steps.add(b -> b.filter(filterExpression)); + return this; + } + + public CallAsyncAPISpec subscription( + Consumer> foreachConfigurer) { + steps.add(b -> b.subscription(foreachConfigurer)); + return this; + } + + public CallAsyncAPISpec authentication(AuthenticationConfigurer authenticationConfigurer) { + steps.add(b -> b.authentication(authenticationConfigurer)); + return this; + } + + @Override + public void accept(CallAsyncAPITaskBuilder builder) { + for (var s : steps) { + s.accept(builder); + } + } +} diff --git a/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/dsl/DSL.java b/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/dsl/DSL.java index c88f9a2d1..671714fc2 100644 --- a/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/dsl/DSL.java +++ b/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/dsl/DSL.java @@ -28,6 +28,7 @@ import io.serverlessworkflow.fluent.spec.TimeoutBuilder; import io.serverlessworkflow.fluent.spec.TryTaskBuilder; import io.serverlessworkflow.fluent.spec.configurers.AuthenticationConfigurer; +import io.serverlessworkflow.fluent.spec.configurers.CallAsyncAPIConfigurer; import io.serverlessworkflow.fluent.spec.configurers.CallGrpcConfigurer; import io.serverlessworkflow.fluent.spec.configurers.CallHttpConfigurer; import io.serverlessworkflow.fluent.spec.configurers.CallOpenAPIConfigurer; @@ -114,6 +115,28 @@ public static CallGrpcSpec grpc() { return new CallGrpcSpec(); } + /** + * Create a new AsyncAPI call specification to be used with {@link #call(CallAsyncAPIConfigurer)}. + * + *

Typical usage: + * + * 

{@code
+   * tasks(
+   *   call(
+   *     asyncapi()
+   *       .document("http://acme.org/asyncapi.yaml")
+   *       .operation("greet")
+   *       .message(Map.of("greeting", "hello"))
+   *   )
+   * );
+   * }
+ * + * @return a new {@link CallAsyncAPISpec} instance + */ + public static CallAsyncAPISpec asyncapi() { + return new CallAsyncAPISpec(); + } + public static WorkflowSpec workflow(String namespace, String name, String version) { return new WorkflowSpec().namespace(namespace).name(name).version(version); } @@ -760,6 +783,27 @@ public static TasksConfigurer call(String name, CallOpenAPIConfigurer configurer return list -> list.openapi(name, configurer); } + /** + * Create a {@link TasksConfigurer} that adds an AsyncAPI call task. + * + * @param configurer AsyncAPI configurer + * @return a {@link TasksConfigurer} that adds a CallAsyncAPI task + */ + public static TasksConfigurer call(CallAsyncAPIConfigurer configurer) { + return list -> list.asyncapi(configurer); + } + + /** + * Create a {@link TasksConfigurer} that adds an AsyncAPI call task with an explicit name. + * + * @param name the task name + * @param configurer AsyncAPI configurer + * @return a {@link TasksConfigurer} that adds a CallAsyncAPI task + */ + public static TasksConfigurer call(String name, CallAsyncAPIConfigurer configurer) { + return list -> list.asyncapi(name, configurer); + } + public static TasksConfigurer call(CallGrpcConfigurer configurer) { return list -> list.grpc(configurer); } diff --git a/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/spi/CallAsyncAPIFluent.java b/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/spi/CallAsyncAPIFluent.java new file mode 100644 index 000000000..0d186f7de --- /dev/null +++ b/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/spi/CallAsyncAPIFluent.java @@ -0,0 +1,28 @@ +/* + * 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.fluent.spec.spi; + +import io.serverlessworkflow.fluent.spec.TaskBaseBuilder; +import java.util.function.Consumer; + +public interface CallAsyncAPIFluent, LIST> { + + LIST asyncapi(String name, Consumer itemsConfigurer); + + default LIST asyncapi(Consumer itemsConfigurer) { + return this.asyncapi(null, itemsConfigurer); + } +} diff --git a/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/spi/CallAsyncAPITaskFluent.java b/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/spi/CallAsyncAPITaskFluent.java new file mode 100644 index 000000000..a8c5aa447 --- /dev/null +++ b/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/spi/CallAsyncAPITaskFluent.java @@ -0,0 +1,244 @@ +/* + * 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.fluent.spec.spi; + +import io.serverlessworkflow.api.types.AsyncApiArguments; +import io.serverlessworkflow.api.types.AsyncApiMessageConsumptionPolicyAmount; +import io.serverlessworkflow.api.types.AsyncApiMessageConsumptionPolicyUnion; +import io.serverlessworkflow.api.types.AsyncApiMessageConsumptionPolicyUntil; +import io.serverlessworkflow.api.types.AsyncApiMessageConsumptionPolicyWhile; +import io.serverlessworkflow.api.types.AsyncApiMessageHeaders; +import io.serverlessworkflow.api.types.AsyncApiMessagePayload; +import io.serverlessworkflow.api.types.AsyncApiOutboundMessage; +import io.serverlessworkflow.api.types.AsyncApiServer; +import io.serverlessworkflow.api.types.AsyncApiSubscription; +import io.serverlessworkflow.api.types.CallAsyncAPI; +import io.serverlessworkflow.api.types.Endpoint; +import io.serverlessworkflow.api.types.EndpointConfiguration; +import io.serverlessworkflow.api.types.EndpointUri; +import io.serverlessworkflow.api.types.ExternalResource; +import io.serverlessworkflow.api.types.ReferenceableAuthenticationPolicy; +import io.serverlessworkflow.api.types.UriTemplate; +import io.serverlessworkflow.fluent.spec.ReferenceableAuthenticationPolicyBuilder; +import io.serverlessworkflow.fluent.spec.SubscriptionIteratorBuilder; +import io.serverlessworkflow.fluent.spec.TaskBaseBuilder; +import io.serverlessworkflow.fluent.spec.TaskItemListBuilder; +import io.serverlessworkflow.fluent.spec.configurers.AuthenticationConfigurer; +import java.net.URI; +import java.util.Map; +import java.util.function.Consumer; + +public interface CallAsyncAPITaskFluent> { + + default CallAsyncAPI build() { + return ((CallAsyncAPI) this.self().getTask()); + } + + SELF self(); + + default SELF document(String uri) { + ((CallAsyncAPI) this.self().getTask()) + .getWith() + .setDocument(new ExternalResource().withEndpoint(EndpointUtil.fromString(uri))); + return self(); + } + + default SELF document(URI uri) { + ((CallAsyncAPI) this.self().getTask()) + .getWith() + .withDocument( + new ExternalResource() + .withEndpoint( + new Endpoint().withUriTemplate(new UriTemplate().withLiteralUri(uri)))); + return self(); + } + + default SELF document(String uri, AuthenticationConfigurer authenticationConfigurer) { + final ReferenceableAuthenticationPolicyBuilder policy = + new ReferenceableAuthenticationPolicyBuilder(); + authenticationConfigurer.accept(policy); + ReferenceableAuthenticationPolicy auth = policy.build(); + ((CallAsyncAPI) this.self().getTask()).getWith().setAuthentication(auth); + ((CallAsyncAPI) this.self().getTask()) + .getWith() + .setDocument(new ExternalResource().withEndpoint(EndpointUtil.fromString(uri, auth))); + return self(); + } + + default SELF document(URI uri, AuthenticationConfigurer authenticationConfigurer) { + final ReferenceableAuthenticationPolicyBuilder policy = + new ReferenceableAuthenticationPolicyBuilder(); + authenticationConfigurer.accept(policy); + ReferenceableAuthenticationPolicy auth = policy.build(); + ((CallAsyncAPI) this.self().getTask()).getWith().setAuthentication(auth); + ((CallAsyncAPI) this.self().getTask()) + .getWith() + .setDocument( + new ExternalResource() + .withEndpoint( + new Endpoint() + .withEndpointConfiguration( + new EndpointConfiguration() + .withUri( + new EndpointUri() + .withLiteralEndpointURI( + new UriTemplate().withLiteralUri(uri))) + .withAuthentication(auth)))); + return self(); + } + + default SELF channel(String channel) { + ((CallAsyncAPI) this.self().getTask()).getWith().setChannel(channel); + return self(); + } + + default SELF operation(String operation) { + ((CallAsyncAPI) this.self().getTask()).getWith().setOperation(operation); + return self(); + } + + default SELF server(String name) { + ((CallAsyncAPI) this.self().getTask()).getWith().setServer(new AsyncApiServer(name)); + return self(); + } + + default SELF server(String name, Map variables) { + AsyncApiServer server = new AsyncApiServer(name); + io.serverlessworkflow.api.types.AsyncApiServerVariables vars = + new io.serverlessworkflow.api.types.AsyncApiServerVariables(); + variables.forEach(vars::withAdditionalProperty); + server.setVariables(vars); + ((CallAsyncAPI) this.self().getTask()).getWith().setServer(server); + return self(); + } + + default SELF protocol(AsyncApiArguments.AsyncApiProtocol protocol) { + ((CallAsyncAPI) this.self().getTask()).getWith().setProtocol(protocol); + return self(); + } + + default SELF message(Map payload) { + AsyncApiOutboundMessage msg = ensureMessage(); + AsyncApiMessagePayload p = new AsyncApiMessagePayload(); + payload.forEach(p::withAdditionalProperty); + msg.setPayload(p); + return self(); + } + + default SELF message(Map payload, Map headers) { + AsyncApiOutboundMessage msg = ensureMessage(); + AsyncApiMessagePayload p = new AsyncApiMessagePayload(); + payload.forEach(p::withAdditionalProperty); + msg.setPayload(p); + AsyncApiMessageHeaders h = new AsyncApiMessageHeaders(); + headers.forEach(h::withAdditionalProperty); + msg.setHeaders(h); + return self(); + } + + default SELF payload(Map payload) { + AsyncApiOutboundMessage msg = ensureMessage(); + AsyncApiMessagePayload p = new AsyncApiMessagePayload(); + payload.forEach(p::withAdditionalProperty); + msg.setPayload(p); + return self(); + } + + default SELF headers(Map headers) { + AsyncApiOutboundMessage msg = ensureMessage(); + AsyncApiMessageHeaders h = new AsyncApiMessageHeaders(); + headers.forEach(h::withAdditionalProperty); + msg.setHeaders(h); + return self(); + } + + private AsyncApiOutboundMessage ensureMessage() { + AsyncApiArguments args = ((CallAsyncAPI) this.self().getTask()).getWith(); + if (args.getMessage() == null) { + args.setMessage(new AsyncApiOutboundMessage()); + } + return args.getMessage(); + } + + default SELF subscription( + Consumer> foreachConfigurer) { + AsyncApiArguments args = ((CallAsyncAPI) this.self().getTask()).getWith(); + if (args.getSubscription() == null) { + args.setSubscription( + new AsyncApiSubscription( + new AsyncApiMessageConsumptionPolicyUnion() + .withAsyncApiMessageConsumptionPolicyAmount( + new AsyncApiMessageConsumptionPolicyAmount(1)))); + } + SubscriptionIteratorBuilder builder = + new SubscriptionIteratorBuilder<>(new TaskItemListBuilder(0)); + foreachConfigurer.accept(builder); + args.getSubscription().setForeach(builder.build()); + return self(); + } + + default SELF consumeAmount(int amount) { + ensureSubscription() + .setConsume( + new AsyncApiMessageConsumptionPolicyUnion() + .withAsyncApiMessageConsumptionPolicyAmount( + new AsyncApiMessageConsumptionPolicyAmount(amount))); + return self(); + } + + default SELF consumeWhile(String expression) { + ensureSubscription() + .setConsume( + new AsyncApiMessageConsumptionPolicyUnion() + .withAsyncApiMessageConsumptionPolicyWhile( + new AsyncApiMessageConsumptionPolicyWhile().withWhile(expression))); + return self(); + } + + default SELF consumeUntil(String expression) { + ensureSubscription() + .setConsume( + new AsyncApiMessageConsumptionPolicyUnion() + .withAsyncApiMessageConsumptionPolicyUntil( + new AsyncApiMessageConsumptionPolicyUntil().withUntil(expression))); + return self(); + } + + default SELF filter(String filterExpression) { + ensureSubscription().setFilter(filterExpression); + return self(); + } + + private AsyncApiSubscription ensureSubscription() { + AsyncApiArguments args = ((CallAsyncAPI) this.self().getTask()).getWith(); + if (args.getSubscription() == null) { + args.setSubscription( + new AsyncApiSubscription( + new AsyncApiMessageConsumptionPolicyUnion() + .withAsyncApiMessageConsumptionPolicyAmount( + new AsyncApiMessageConsumptionPolicyAmount(1)))); + } + return args.getSubscription(); + } + + default SELF authentication(AuthenticationConfigurer authenticationConfigurer) { + final ReferenceableAuthenticationPolicyBuilder policy = + new ReferenceableAuthenticationPolicyBuilder(); + authenticationConfigurer.accept(policy); + ((CallAsyncAPI) this.self().getTask()).getWith().setAuthentication(policy.build()); + return self(); + } +} diff --git a/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/spi/DoFluent.java b/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/spi/DoFluent.java index 37c5f461b..249e3a1c3 100644 --- a/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/spi/DoFluent.java +++ b/fluent/spec/src/main/java/io/serverlessworkflow/fluent/spec/spi/DoFluent.java @@ -15,6 +15,7 @@ */ package io.serverlessworkflow.fluent.spec.spi; +import io.serverlessworkflow.fluent.spec.CallAsyncAPITaskBuilder; import io.serverlessworkflow.fluent.spec.CallGrpcTaskBuilder; import io.serverlessworkflow.fluent.spec.CallHttpTaskBuilder; import io.serverlessworkflow.fluent.spec.CallOpenAPITaskBuilder; @@ -49,5 +50,6 @@ public interface DoFluent WaitFluent, RaiseFluent, CallOpenAPIFluent, + CallAsyncAPIFluent, CallGrpcFluent, WorkflowFluent {} diff --git a/fluent/spec/src/test/java/io/serverlessworkflow/fluent/spec/WorkflowBuilderTest.java b/fluent/spec/src/test/java/io/serverlessworkflow/fluent/spec/WorkflowBuilderTest.java index 4ba41c203..d12290c42 100644 --- a/fluent/spec/src/test/java/io/serverlessworkflow/fluent/spec/WorkflowBuilderTest.java +++ b/fluent/spec/src/test/java/io/serverlessworkflow/fluent/spec/WorkflowBuilderTest.java @@ -33,6 +33,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertInstanceOf; import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import io.serverlessworkflow.api.types.AuthenticationPolicyUnion; @@ -112,6 +113,17 @@ void testUseAuthenticationsBasic() { assertNotNull(union.getBasicAuthenticationPolicy(), "BasicAuthenticationPolicy should be set"); } + @Test + void testEmptyTasksThrows() { + assertThrows(IllegalStateException.class, () -> WorkflowBuilder.workflow().tasks().build()); + } + + @Test + void testEmptyTasksConsumerThrows() { + assertThrows( + IllegalStateException.class, () -> WorkflowBuilder.workflow().tasks(d -> {}).build()); + } + @Test void testDoTaskSetAndForEach() { Workflow wf = diff --git a/fluent/spec/src/test/java/io/serverlessworkflow/fluent/spec/dsl/CallAsyncApiDslTest.java b/fluent/spec/src/test/java/io/serverlessworkflow/fluent/spec/dsl/CallAsyncApiDslTest.java new file mode 100644 index 000000000..960b41047 --- /dev/null +++ b/fluent/spec/src/test/java/io/serverlessworkflow/fluent/spec/dsl/CallAsyncApiDslTest.java @@ -0,0 +1,210 @@ +/* + * 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.fluent.spec.dsl; + +import static io.serverlessworkflow.fluent.spec.dsl.DSL.asyncapi; +import static io.serverlessworkflow.fluent.spec.dsl.DSL.basic; +import static io.serverlessworkflow.fluent.spec.dsl.DSL.call; +import static org.assertj.core.api.Assertions.assertThat; + +import io.serverlessworkflow.api.types.AsyncApiArguments; +import io.serverlessworkflow.api.types.Workflow; +import io.serverlessworkflow.fluent.spec.WorkflowBuilder; +import java.util.Map; +import org.junit.jupiter.api.Test; + +public class CallAsyncApiDslTest { + + @Test + void when_call_asyncapi_publish_with_message() { + Workflow wf = + WorkflowBuilder.workflow("f", "ns", "1") + .tasks( + call( + asyncapi() + .document("https://example.com/asyncapi.yaml") + .operation("greet") + .message( + Map.of("greeting", "${ .name }"), + Map.of("content-type", "application/json")))) + .build(); + + var taskItem = wf.getDo().get(0); + var callAsyncAPI = taskItem.getTask().getCallTask().getCallAsyncAPI(); + assertThat(callAsyncAPI).isNotNull(); + + var with = callAsyncAPI.getWith(); + assertThat(with).isNotNull(); + assertThat(with.getDocument()).isNotNull(); + assertThat(with.getOperation()).isEqualTo("greet"); + + assertThat(with.getMessage()).isNotNull(); + assertThat(with.getMessage().getPayload()).isNotNull(); + assertThat(with.getMessage().getPayload().getAdditionalProperties()) + .containsEntry("greeting", "${ .name }"); + assertThat(with.getMessage().getHeaders()).isNotNull(); + assertThat(with.getMessage().getHeaders().getAdditionalProperties()) + .containsEntry("content-type", "application/json"); + } + + @Test + void when_call_asyncapi_subscribe_with_amount() { + Workflow wf = + WorkflowBuilder.workflow("f", "ns", "1") + .tasks( + call( + asyncapi() + .document("https://example.com/asyncapi.yaml") + .operation("receive") + .consumeAmount(5))) + .build(); + + var with = wf.getDo().get(0).getTask().getCallTask().getCallAsyncAPI().getWith(); + assertThat(with.getSubscription()).isNotNull(); + assertThat(with.getSubscription().getConsume()).isNotNull(); + assertThat( + with.getSubscription() + .getConsume() + .getAsyncApiMessageConsumptionPolicyAmount() + .getAmount()) + .isEqualTo(5); + } + + @Test + void when_call_asyncapi_subscribe_with_until() { + Workflow wf = + WorkflowBuilder.workflow("f", "ns", "1") + .tasks( + call( + asyncapi() + .document("https://example.com/asyncapi.yaml") + .operation("receive") + .consumeUntil("${ (. | length) >= 2 }"))) + .build(); + + var with = wf.getDo().get(0).getTask().getCallTask().getCallAsyncAPI().getWith(); + assertThat(with.getSubscription()).isNotNull(); + assertThat( + with.getSubscription() + .getConsume() + .getAsyncApiMessageConsumptionPolicyUntil() + .getUntil()) + .isEqualTo("${ (. | length) >= 2 }"); + } + + @Test + void when_call_asyncapi_with_channel_and_protocol() { + Workflow wf = + WorkflowBuilder.workflow("f", "ns", "1") + .tasks( + call( + asyncapi() + .document("https://example.com/asyncapi.yaml") + .channel("greetings") + .protocol(AsyncApiArguments.AsyncApiProtocol.KAFKA) + .message(Map.of("hello", "world")))) + .build(); + + var with = wf.getDo().get(0).getTask().getCallTask().getCallAsyncAPI().getWith(); + assertThat(with.getChannel()).isEqualTo("greetings"); + assertThat(with.getProtocol()).isEqualTo(AsyncApiArguments.AsyncApiProtocol.KAFKA); + } + + @Test + void when_call_asyncapi_with_explicit_name() { + Workflow wf = + WorkflowBuilder.workflow("f", "ns", "1") + .tasks( + call( + "myAsyncCall", + asyncapi() + .document("https://example.com/asyncapi.yaml") + .operation("greet") + .message(Map.of("greeting", "hello")))) + .build(); + + assertThat(wf.getDo()).hasSize(1); + assertThat(wf.getDo().get(0).getName()).isEqualTo("myAsyncCall"); + assertThat(wf.getDo().get(0).getTask().getCallTask().getCallAsyncAPI()).isNotNull(); + } + + @Test + void when_call_asyncapi_with_server() { + Workflow wf = + WorkflowBuilder.workflow("f", "ns", "1") + .tasks( + call( + asyncapi() + .document("https://example.com/asyncapi.yaml") + .operation("greet") + .server("production") + .message(Map.of("greeting", "hello")))) + .build(); + + var with = wf.getDo().get(0).getTask().getCallTask().getCallAsyncAPI().getWith(); + assertThat(with.getServer()).isNotNull(); + assertThat(with.getServer().getName()).isEqualTo("production"); + } + + @Test + void when_call_asyncapi_with_basic_auth_on_document() { + Workflow wf = + WorkflowBuilder.workflow("f", "ns", "1") + .tasks( + call( + asyncapi() + .document("https://example.com/asyncapi.yaml", basic("alice", "secret")) + .operation("greet") + .message(Map.of("greeting", "hello")))) + .build(); + + var with = wf.getDo().get(0).getTask().getCallTask().getCallAsyncAPI().getWith(); + assertThat(with.getAuthentication()).isNotNull(); + assertThat(with.getAuthentication().getAuthenticationPolicy()).isNotNull(); + assertThat( + with.getAuthentication() + .getAuthenticationPolicy() + .getBasicAuthenticationPolicy() + .getBasic() + .getBasicAuthenticationProperties() + .getUsername()) + .isEqualTo("alice"); + } + + @Test + void when_call_asyncapi_with_filter() { + Workflow wf = + WorkflowBuilder.workflow("f", "ns", "1") + .tasks( + call( + asyncapi() + .document("https://example.com/asyncapi.yaml") + .operation("receive") + .filter("${ .payload.roomId == \"room-1\" }") + .consumeAmount(2))) + .build(); + + var with = wf.getDo().get(0).getTask().getCallTask().getCallAsyncAPI().getWith(); + assertThat(with.getSubscription()).isNotNull(); + assertThat(with.getSubscription().getFilter()).isEqualTo("${ .payload.roomId == \"room-1\" }"); + assertThat( + with.getSubscription() + .getConsume() + .getAsyncApiMessageConsumptionPolicyAmount() + .getAmount()) + .isEqualTo(2); + } +} 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 d125572cd..e393dcda0 100644 --- a/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowApplication.java +++ b/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowApplication.java @@ -114,6 +114,7 @@ public class WorkflowApplication implements AutoCloseable { private final WorkflowLifeCycleCloudEventFactory lifeCycleCloudEventFactory; private final ScheduledExecutorService schedulerExecutorService; private final Set allowedCommands; + private final Map, ServiceLoader> servicesLoaded = new ConcurrentHashMap<>(); private WorkflowApplication(Builder builder) { this.taskFactory = builder.taskFactory; @@ -708,4 +709,22 @@ public WorkflowLifeCycleCloudEventFactory lifeCycleCloudEventFactory() { public Set allowedCommands() { return allowedCommands; } + + @SuppressWarnings("unchecked") + public > List serviceLoadedClasses(Class clazz) { + ServiceLoader serviceLoader = servicesLoaded.computeIfAbsent(clazz, ServiceLoader::load); + return (List) serviceLoader.stream().map(ServiceLoader.Provider::get).sorted().toList(); + } + + public > T serviceLoadedClass(Class serviceClass) { + ServiceLoader serviceLoader = + servicesLoaded.computeIfAbsent(serviceClass, ServiceLoader::load); + return (T) + serviceLoader.stream() + .map(ServiceLoader.Provider::get) + .sorted() + .findFirst() + .orElseThrow( + () -> new IllegalStateException("No " + serviceClass + " implementation found")); + } } diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/auth/CommonOAuthProvider.java b/impl/core/src/main/java/io/serverlessworkflow/impl/auth/CommonOAuthProvider.java index e1e8fb6e2..3404ac1d2 100644 --- a/impl/core/src/main/java/io/serverlessworkflow/impl/auth/CommonOAuthProvider.java +++ b/impl/core/src/main/java/io/serverlessworkflow/impl/auth/CommonOAuthProvider.java @@ -16,13 +16,13 @@ package io.serverlessworkflow.impl.auth; import static io.serverlessworkflow.impl.WorkflowUtils.checkSecret; -import static io.serverlessworkflow.impl.WorkflowUtils.loadFirst; import static io.serverlessworkflow.impl.WorkflowUtils.secret; import io.serverlessworkflow.api.types.OAuth2AuthenticationData; import io.serverlessworkflow.api.types.SecretBasedAuthenticationPolicy; import io.serverlessworkflow.api.types.Workflow; import io.serverlessworkflow.impl.TaskContext; +import io.serverlessworkflow.impl.WorkflowApplication; import io.serverlessworkflow.impl.WorkflowContext; import io.serverlessworkflow.impl.WorkflowModel; import io.serverlessworkflow.impl.WorkflowValueResolver; @@ -35,14 +35,6 @@ public abstract class CommonOAuthProvider implements AuthProvider { private final WorkflowValueResolver tokenProvider; - private static JWTConverter jwtConverter = - loadFirst(JWTConverter.class) - .orElseThrow(() -> new IllegalStateException("No JWTConverter implementation found")); - - private static AccessTokenProviderFactory accessTokenProviderFactory = - loadFirst(AccessTokenProviderFactory.class) - .orElseThrow(() -> new IllegalStateException("No JWTConverter implementation found")); - protected CommonOAuthProvider(WorkflowValueResolver tokenProvider) { this.tokenProvider = tokenProvider; } @@ -67,35 +59,42 @@ protected static OAuth2AuthenticationData fillFromMap( } protected static WorkflowValueResolver accessToken( + WorkflowApplication app, Workflow workflow, OAuth2AuthenticationData authenticationData, SecretBasedAuthenticationPolicy secret, AuthRequestBuilder builder) { if (authenticationData != null) { - return build(authenticationData, builder); + return build(authenticationData, builder, app); } else if (secret != null) { - return build(checkSecret(workflow, secret), builder); + return build(checkSecret(workflow, secret), builder, app); } throw new IllegalStateException("Both policy and secret are null"); } private static WorkflowValueResolver build( - OAuth2AuthenticationData authenticationData, AuthRequestBuilder authBuilder) { + OAuth2AuthenticationData authenticationData, + AuthRequestBuilder authBuilder, + WorkflowApplication app) { AccessTokenProvider tokenProvider = - accessTokenProviderFactory.build( - authBuilder.apply(authenticationData), authenticationData.getIssuers(), jwtConverter); + app.serviceLoadedClass(AccessTokenProviderFactory.class) + .build( + authBuilder.apply(authenticationData), + authenticationData.getIssuers(), + app.serviceLoadedClass(JWTConverter.class)); return (w, t, m) -> tokenProvider; } private static WorkflowValueResolver build( - String secretName, AuthRequestBuilder authBuilder) { + String secretName, AuthRequestBuilder authBuilder, WorkflowApplication app) { return (w, t, m) -> { Map secret = secret(w, secretName); String issuers = (String) secret.get("issuers"); - return accessTokenProviderFactory.build( - authBuilder.apply(secret), - issuers != null ? Arrays.asList(issuers.split(",")) : null, - jwtConverter); + return app.serviceLoadedClass(AccessTokenProviderFactory.class) + .build( + authBuilder.apply(secret), + issuers != null ? Arrays.asList(issuers.split(",")) : null, + app.serviceLoadedClass(JWTConverter.class)); }; } } diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/auth/OAuth2AuthProvider.java b/impl/core/src/main/java/io/serverlessworkflow/impl/auth/OAuth2AuthProvider.java index 6bce3d814..16e0e1ea7 100644 --- a/impl/core/src/main/java/io/serverlessworkflow/impl/auth/OAuth2AuthProvider.java +++ b/impl/core/src/main/java/io/serverlessworkflow/impl/auth/OAuth2AuthProvider.java @@ -24,6 +24,7 @@ public OAuth2AuthProvider( WorkflowApplication application, Workflow workflow, OAuthPolicyData policyData) { super( accessToken( + application, workflow, policyData.data(), policyData.secret(), diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/auth/OpenIdAuthProvider.java b/impl/core/src/main/java/io/serverlessworkflow/impl/auth/OpenIdAuthProvider.java index 80dd4138b..425e3c082 100644 --- a/impl/core/src/main/java/io/serverlessworkflow/impl/auth/OpenIdAuthProvider.java +++ b/impl/core/src/main/java/io/serverlessworkflow/impl/auth/OpenIdAuthProvider.java @@ -24,6 +24,7 @@ public OpenIdAuthProvider( WorkflowApplication application, Workflow workflow, OAuthPolicyData policyData) { super( accessToken( + application, workflow, policyData.data(), policyData.secret(), diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/DefaultTaskExecutorFactory.java b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/DefaultTaskExecutorFactory.java index 55363ac99..85a198d9b 100644 --- a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/DefaultTaskExecutorFactory.java +++ b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/DefaultTaskExecutorFactory.java @@ -18,6 +18,7 @@ import io.serverlessworkflow.api.types.CallTask; import io.serverlessworkflow.api.types.Task; import io.serverlessworkflow.api.types.TaskBase; +import io.serverlessworkflow.impl.WorkflowApplication; import io.serverlessworkflow.impl.WorkflowDefinition; import io.serverlessworkflow.impl.WorkflowMutablePosition; import io.serverlessworkflow.impl.executors.CallTaskExecutor.CallTaskExecutorBuilder; @@ -32,9 +33,7 @@ import io.serverlessworkflow.impl.executors.SwitchExecutor.SwitchExecutorBuilder; import io.serverlessworkflow.impl.executors.TryExecutor.TryExecutorBuilder; import io.serverlessworkflow.impl.executors.WaitExecutor.WaitExecutorBuilder; -import java.util.Collection; -import java.util.ServiceLoader; -import java.util.ServiceLoader.Provider; +import java.util.List; public class DefaultTaskExecutorFactory implements TaskExecutorFactory { @@ -46,9 +45,6 @@ public static TaskExecutorFactory get() { protected DefaultTaskExecutorFactory() {} - private Collection callTasks = - ServiceLoader.load(CallableTaskBuilder.class).stream().map(Provider::get).sorted().toList(); - @Override public TaskExecutorBuilder getTaskExecutor( WorkflowMutablePosition position, Task task, WorkflowDefinition definition) { @@ -57,7 +53,10 @@ public TaskExecutorBuilder getTaskExecutor( TaskBase taskBase = (TaskBase) callTask.get(); if (taskBase != null) { return new CallTaskExecutorBuilder( - position, taskBase, definition, findCallTask(taskBase.getClass())); + position, + taskBase, + definition, + findCallTask(taskBase.getClass(), definition.application())); } } else if (task.getSwitchTask() != null) { return new SwitchExecutorBuilder(position, task.getSwitchTask(), definition); @@ -86,7 +85,9 @@ public TaskExecutorBuilder getTaskExecutor( } @SuppressWarnings("unchecked") - private CallableTaskBuilder findCallTask(Class clazz) { + private CallableTaskBuilder findCallTask( + Class clazz, WorkflowApplication app) { + List callTasks = app.serviceLoadedClasses(CallableTaskBuilder.class); return (CallableTaskBuilder) callTasks.stream() .filter(s -> s.accept(clazz)) 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 f34253471..1d42b9b5d 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 @@ -41,16 +41,10 @@ import java.util.Collection; import java.util.Map; import java.util.Optional; -import java.util.ServiceLoader; import java.util.concurrent.CompletableFuture; public class EmitExecutor extends RegularTaskExecutor { - private static final Collection emittedDecorators = - ServiceLoader.load(EmittedEventDecorator.class).stream() - .map(ServiceLoader.Provider::get) - .sorted() - .toList(); private final EventPropertiesBuilder props; public static class EmitExecutorBuilder @@ -139,7 +133,11 @@ private CloudEvent buildCloudEvent(WorkflowContext workflow, TaskContext taskCon .additionalFilter() .map(filter -> filter.apply(workflow, taskContext, taskContext.input())) .ifPresent(value -> value.forEach((k, v) -> addExtension(ceBuilder, k, v))); - emittedDecorators.forEach(d -> d.decorate(ceBuilder, workflow, taskContext)); + workflow + .definition() + .application() + .serviceLoadedClasses(EmittedEventDecorator.class) + .forEach(d -> d.decorate(ceBuilder, workflow, taskContext)); return ceBuilder.build(); } diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/RunScriptExecutorBuilder.java b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/RunScriptExecutorBuilder.java index 87198d540..02d2b813b 100644 --- a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/RunScriptExecutorBuilder.java +++ b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/RunScriptExecutorBuilder.java @@ -27,7 +27,6 @@ import io.serverlessworkflow.impl.scripts.ScriptRunner; import java.util.Objects; import java.util.Optional; -import java.util.ServiceLoader; public class RunScriptExecutorBuilder implements RunnableTaskBuilder { @@ -65,10 +64,8 @@ public CallableTask build(RunScript taskConfiguration, WorkflowDefinition defini m), taskConfiguration.isAwait(), taskConfiguration.getReturn(), - ServiceLoader.load(ScriptRunner.class).stream() - .map(ServiceLoader.Provider::get) + application.serviceLoadedClasses(ScriptRunner.class).stream() .filter(s -> s.identifier().equals(language)) - .sorted() .findFirst() .orElseThrow( () -> diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/RunTaskExecutor.java b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/RunTaskExecutor.java index c398a36d7..f1fe7861e 100644 --- a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/RunTaskExecutor.java +++ b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/RunTaskExecutor.java @@ -22,17 +22,12 @@ import io.serverlessworkflow.impl.WorkflowDefinition; import io.serverlessworkflow.impl.WorkflowModel; import io.serverlessworkflow.impl.WorkflowMutablePosition; -import java.util.ServiceLoader; -import java.util.ServiceLoader.Provider; import java.util.concurrent.CompletableFuture; public class RunTaskExecutor extends RegularTaskExecutor { private final CallableTask runnable; - private static final ServiceLoader runnables = - ServiceLoader.load(RunnableTaskBuilder.class); - public static class RunTaskExecutorBuilder extends RegularTaskExecutorBuilder { private CallableTask runnable; @@ -42,10 +37,8 @@ protected RunTaskExecutorBuilder( super(position, task, definition); RunTaskConfiguration config = task.getRun().get(); this.runnable = - runnables.stream() - .map(Provider::get) + definition.application().serviceLoadedClasses(RunnableTaskBuilder.class).stream() .filter(r -> r.accept(config.getClass())) - .sorted() .findFirst() .map(r -> r.build(config, definition)) .orElseThrow( diff --git a/impl/http/src/main/java/io/serverlessworkflow/impl/executors/http/HttpExecutorBuilder.java b/impl/http/src/main/java/io/serverlessworkflow/impl/executors/http/HttpExecutorBuilder.java index 5c9f025bb..50693a5ee 100644 --- a/impl/http/src/main/java/io/serverlessworkflow/impl/executors/http/HttpExecutorBuilder.java +++ b/impl/http/src/main/java/io/serverlessworkflow/impl/executors/http/HttpExecutorBuilder.java @@ -30,13 +30,12 @@ import java.util.List; import java.util.Map; import java.util.Optional; -import java.util.ServiceLoader; public class HttpExecutorBuilder { public static final String HTTP_REQUEST_DECORATOR_KEY = "HttpRequestDecorators"; private final WorkflowDefinition definition; - private final List requestDecorators; + private List requestDecorators; private WorkflowValueResolver pathSupplier; private Object body; private String method = HttpMethod.GET; @@ -47,13 +46,13 @@ public class HttpExecutorBuilder { private HttpExecutorBuilder(WorkflowDefinition definition) { this.definition = definition; - this.requestDecorators = new ArrayList<>(); + this.requestDecorators = + new ArrayList<>(definition.application().serviceLoadedClasses(HttpRequestDecorator.class)); requestDecorators.addAll( definition .application() .>additionalObject(HTTP_REQUEST_DECORATOR_KEY) .orElse(List.of())); - ServiceLoader.load(HttpRequestDecorator.class).forEach(requestDecorators::add); Collections.sort(requestDecorators); }