From e0629b5080149060c3e6af39711fd094421d1058 Mon Sep 17 00:00:00 2001 From: abilan Date: Mon, 8 May 2023 17:52:21 -0400 Subject: [PATCH] Introducing `spring-cloud-function-integration` Spring Integration Java DSL, is a tool to compose integration flows programmatically. We can build the flow not only based on standard EIP components, but also using protocol-specific channel adapters. Any generic services also can be used as handler in the flow. This includes simple lambda operations or functions. On the other hand Spring Cloud Function provides a `FunctionCatalog` for registered functions and their compositions & conversions. With this change we introduce a more high-level DSL to use functions from catalog directly in the `IntegrationFlow` to gain the best from both worlds. * Introduce `spring-cloud-function-integration` module based on `spring-cloud-function-context` and `spring-boot-starter-integration` * Expose a `FunctionFlowBuilder` auto-configuration * Add `FunctionFlowDefinition` to expose `apply()` and `accept()` operators * Document this new module --- docs/src/main/asciidoc/index.adoc | 3 +- .../src/main/asciidoc/spring-integration.adoc | 101 ++++++++++ pom.xml | 1 + spring-cloud-function-integration/pom.xml | 41 +++++ .../dsl/FunctionFlowAutoConfiguration.java | 40 ++++ .../integration/dsl/FunctionFlowBuilder.java | 172 ++++++++++++++++++ .../dsl/FunctionFlowDefinition.java | 94 ++++++++++ .../integration/dsl/FunctionLookupHelper.java | 86 +++++++++ ...ot.autoconfigure.AutoConfiguration.imports | 1 + .../integration/dsl/FunctionFlowTests.java | 136 ++++++++++++++ 10 files changed, 674 insertions(+), 1 deletion(-) create mode 100644 docs/src/main/asciidoc/spring-integration.adoc create mode 100644 spring-cloud-function-integration/pom.xml create mode 100644 spring-cloud-function-integration/src/main/java/org/springframework/cloud/function/integration/dsl/FunctionFlowAutoConfiguration.java create mode 100644 spring-cloud-function-integration/src/main/java/org/springframework/cloud/function/integration/dsl/FunctionFlowBuilder.java create mode 100644 spring-cloud-function-integration/src/main/java/org/springframework/cloud/function/integration/dsl/FunctionFlowDefinition.java create mode 100644 spring-cloud-function-integration/src/main/java/org/springframework/cloud/function/integration/dsl/FunctionLookupHelper.java create mode 100644 spring-cloud-function-integration/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports create mode 100644 spring-cloud-function-integration/src/test/java/org/springframework/cloud/function/integration/dsl/FunctionFlowTests.java diff --git a/docs/src/main/asciidoc/index.adoc b/docs/src/main/asciidoc/index.adoc index ff8153e03..ebc517b8a 100644 --- a/docs/src/main/asciidoc/index.adoc +++ b/docs/src/main/asciidoc/index.adoc @@ -1,5 +1,5 @@ = Spring Cloud Function Reference Documentation -Mark Fisher, Dave Syer, Oleg Zhurakousky, Anshul Mehra, Dan Dobrin, Chris Bono +Mark Fisher, Dave Syer, Oleg Zhurakousky, Anshul Mehra, Dan Dobrin, Chris Bono, Artem Bilan *{project-version}* @@ -11,6 +11,7 @@ The reference documentation consists of the following sections: <> :: Spring Cloud Function Reference https://github.com/spring-cloud/spring-cloud-function/tree/master/spring-cloud-function-samples/function-sample-cloudevent[Cloud Events] :: Cloud Events https://github.com/spring-cloud/spring-cloud-function/tree/master/spring-cloud-function-rsocket[RSocket] :: RSocket +<<./spring-integration.adoc#spring-integration,Spring Integration>> :: Spring Integration Framework Interaction <> :: AWS Adapter Reference <> :: Azure Adapter Reference <> :: GCP Adapter Reference diff --git a/docs/src/main/asciidoc/spring-integration.adoc b/docs/src/main/asciidoc/spring-integration.adoc new file mode 100644 index 000000000..1c3edf455 --- /dev/null +++ b/docs/src/main/asciidoc/spring-integration.adoc @@ -0,0 +1,101 @@ +[[spring-integration]] +== Spring Integration Interaction + +https://spring.io/projects/spring-integration[Spring Integration Framework] extends the Spring programming model to support the well-known Enterprise Integration Patterns. +It enables lightweight messaging within Spring-based applications and supports integration with external systems via declarative adapters. +It also provides a high-level DSL to compose various operations (endpoints) into a logical integration flow. +With a lambda style of this DSL configuration, Spring Integration already has a good level of `java.util.function` interfaces adoption. +The `@MessagingGateway` proxy interface can also be as a `Function` or `Consumer`, which according to the Spring Cloud Function environment can be registered into a function catalog. +See more information in Spring Integration https://docs.spring.io/spring-integration/docs/current/reference/html/messaging-endpoints.html#functions-support[ReferenceManual] about its support for functions. + +On the other hand, starting with version `4.0.3`, Spring Cloud Function introduces a `spring-cloud-function-integration` module which provides deeper, more cloud-specific and auto-configuration based API for interaction with a `FunctionCatalog` from Spring Integration DSL perspective. +The `FunctionFlowBuilder` is auto-configured and autowired with a `FunctionCatalog` and represents an entry point for function-specific DSL for target `IntegrationFlow` instance. +In addition to standard `IntegrationFlow.from()` factories (for convenience), the `FunctionFlowBuilder` exposes a `fromSupplier(String supplierDefinition)` factory to lookup the target `Supplier` in the provided `FunctionCatalog`. +Then this `FunctionFlowBuilder` leads to the `FunctionFlowDefinition`. +This `FunctionFlowDefinition` is an implementation of the `IntegrationFlowExtension` and exposes `apply(String functionDefinition)` and `accept(String consumerDefinition)` operators to lookup `Function` or `Consumer` from the `FunctionCatalog`, respectively. +See their Javadocs for more information. + +The following example demonstrates the `FunctionFlowBuilder` in action alongside with the power of the rest of `IntegrationFlow` API: + +[source,java] +---- +@Configuration +static class IntegrationConfiguration { + + @Bean + Supplier simpleByteArraySupplier() { + return "simple test data"::getBytes; + } + + @Bean + Function upperCaseFunction() { + return String::toUpperCase; + } + + @Bean + BlockingQueue results() { + return new LinkedBlockingQueue<>(); + } + + @Bean + Consumer simpleStringConsumer(BlockingQueue results) { + return results::add; + } + + @Bean + QueueChannel wireTapChannel() { + return new QueueChannel(); + } + + @Bean + IntegrationFlow someFunctionFlow(FunctionFlowBuilder functionFlowBuilder) { + return functionFlowBuilder + .fromSupplier("simpleByteArraySupplier") + .wireTap("wireTapChannel") + .apply("upperCaseFunction") + .log(LoggingHandler.Level.WARN) + .accept("simpleStringConsumer"); + } + +} +---- + +Since the `FunctionCatalog.lookup()` functionality is not limited just to simple function names, a function composition feature can also be used in the mentioned `apply()` and `accept()` operators: + +[source,java] +---- +@Bean +IntegrationFlow functionCompositionFlow(FunctionFlowBuilder functionFlowBuilder) { + return functionFlowBuilder + .from("functionCompositionInput") + .accept("upperCaseFunction|simpleStringConsumer"); +} +---- + +This API becomes more relevant, when we add into our Spring Cloud applications auto-configuration dependencies for predefined functions. +For example https://spring.io/projects/spring-cloud-stream-applications[Stream Applications] project, in addition to application images, provides artifacts with functions for various integration use-case, e.g. `debezium-supplier`, `elasticsearch-consumer`, `aggregator-function` etc. + +The following configuration is based on the `http-supplier`, `spel-function` and `file-consumer`, respectively: + +[source,java] +---- +@Bean +IntegrationFlow someFunctionFlow(FunctionFlowBuilder functionFlowBuilder) { + return functionFlowBuilder + .fromSupplier("httpSupplier", e -> e.poller(Pollers.trigger(new OnlyOnceTrigger()))) + .>handle((fluxPayload, headers) -> fluxPayload, e -> e.async(true)) + .channel(c -> c.flux()) + .apply("spelFunction") + .transform(String::toUpperCase) + .accept("fileConsumer"); +} +---- + +What we would need else is just to add their configuration into an `application.properties` (if necessary): + +[source,properties] +---- +http.path-pattern=/testPath +spel.function.expression=new String(payload) +file.consumer.name=test-data.txt +---- diff --git a/pom.xml b/pom.xml index 0f7104b43..816d32135 100644 --- a/pom.xml +++ b/pom.xml @@ -161,6 +161,7 @@ spring-cloud-function-samples spring-cloud-function-deployer spring-cloud-function-adapters + spring-cloud-function-integration spring-cloud-function-rsocket spring-cloud-function-kotlin docs diff --git a/spring-cloud-function-integration/pom.xml b/spring-cloud-function-integration/pom.xml new file mode 100644 index 000000000..761b4a98b --- /dev/null +++ b/spring-cloud-function-integration/pom.xml @@ -0,0 +1,41 @@ + + + + 4.0.0 + + spring-cloud-function-integration + Spring Cloud Function with Spring Integration + Spring Cloud Function with Spring Integration + + + spring-cloud-function-parent + org.springframework.cloud + 4.0.3-SNAPSHOT + + + + + + org.springframework.cloud + spring-cloud-function-context + + + org.springframework.boot + spring-boot-starter-integration + + + + org.springframework.boot + spring-boot-starter-test + test + + + org.springframework.integration + spring-integration-test + test + + + + diff --git a/spring-cloud-function-integration/src/main/java/org/springframework/cloud/function/integration/dsl/FunctionFlowAutoConfiguration.java b/spring-cloud-function-integration/src/main/java/org/springframework/cloud/function/integration/dsl/FunctionFlowAutoConfiguration.java new file mode 100644 index 000000000..03d3848ae --- /dev/null +++ b/spring-cloud-function-integration/src/main/java/org/springframework/cloud/function/integration/dsl/FunctionFlowAutoConfiguration.java @@ -0,0 +1,40 @@ +/* + * Copyright 2023-2023 the original author or 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 + * + * https://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 org.springframework.cloud.function.integration.dsl; + +import org.springframework.boot.autoconfigure.AutoConfiguration; +import org.springframework.cloud.function.context.FunctionCatalog; +import org.springframework.cloud.function.context.config.ContextFunctionCatalogAutoConfiguration; +import org.springframework.context.annotation.Bean; + +/** + * The auto-configuration to expose a {@link FunctionFlowBuilder} bean + * based on the auto-configured {@link FunctionCatalog}. + * + * @author Artem Bilan + * + * @since 4.0.3 + */ +@AutoConfiguration(after = ContextFunctionCatalogAutoConfiguration.class) +public class FunctionFlowAutoConfiguration { + + @Bean + FunctionFlowBuilder functionFlowBuilder(FunctionCatalog functionCatalog) { + return new FunctionFlowBuilder(functionCatalog); + } + +} diff --git a/spring-cloud-function-integration/src/main/java/org/springframework/cloud/function/integration/dsl/FunctionFlowBuilder.java b/spring-cloud-function-integration/src/main/java/org/springframework/cloud/function/integration/dsl/FunctionFlowBuilder.java new file mode 100644 index 000000000..1ad83a440 --- /dev/null +++ b/spring-cloud-function-integration/src/main/java/org/springframework/cloud/function/integration/dsl/FunctionFlowBuilder.java @@ -0,0 +1,172 @@ +/* + * Copyright 2023-2023 the original author or 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 + * + * https://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 org.springframework.cloud.function.integration.dsl; + +import java.util.function.Consumer; +import java.util.function.Supplier; + +import org.reactivestreams.Publisher; + +import org.springframework.cloud.function.context.FunctionCatalog; +import org.springframework.integration.core.MessageSource; +import org.springframework.integration.dsl.GatewayProxySpec; +import org.springframework.integration.dsl.IntegrationFlow; +import org.springframework.integration.dsl.IntegrationFlowBuilder; +import org.springframework.integration.dsl.MessageChannelSpec; +import org.springframework.integration.dsl.MessageProducerSpec; +import org.springframework.integration.dsl.MessageSourceSpec; +import org.springframework.integration.dsl.MessagingGatewaySpec; +import org.springframework.integration.dsl.SourcePollingChannelAdapterSpec; +import org.springframework.integration.endpoint.MessageProducerSupport; +import org.springframework.integration.gateway.MessagingGatewaySupport; +import org.springframework.lang.Nullable; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.util.Assert; + +/** + * The entry point for starting a {@link FunctionFlowDefinition}. + * Requires a {@link FunctionCatalog} to lookup function instances + * by their names or definitions from respective operators. + *

+ * In addition to standard {@link IntegrationFlow} {@code from()} overloaded methods (for convenience), + * this class introduces {@link #fromSupplier(String)} factory methods to resolve the target {@link Supplier} + * by its name or function definition from the provided {@link FunctionCatalog}. + *

+ * This class represents a DSL for functions composition via integration endpoints. + * Extra processing can be done in between functions by the regular {@link IntegrationFlow} operators: + * 

+ * {@code
+ * @Bean
+ * IntegrationFlow someFunctionFlow(FunctionFlowBuilder functionFlowBuilder) {
+ *		return functionFlowBuilder
+ *				.fromSupplier("timeSupplier")
+ *				.apply("spelFunction")
+ *				.log(LoggingHandler.Level.DEBUG, "some.log.category")
+ *				.transform(String::toUpperCase)
+ *				.accept("fileConsumer");
+ * }
+ * }
+ *
+ * + * @author Artem Bilan + * + * @since 4.0.3 + */ +public class FunctionFlowBuilder { + + private final FunctionLookupHelper functionLookupHelper; + + public FunctionFlowBuilder(FunctionCatalog functionCatalog) { + Assert.notNull(functionCatalog, "'functionCatalog' must not be null"); + this.functionLookupHelper = new FunctionLookupHelper(functionCatalog); + } + + public FunctionFlowDefinition fromSupplier(String supplierDefinition) { + return fromSupplier(supplierDefinition, null); + } + + public FunctionFlowDefinition fromSupplier(String supplierDefinition, + @Nullable Consumer endpointConfigurer) { + + return fromSupplier(this.functionLookupHelper.lookupSupplier(supplierDefinition), endpointConfigurer); + } + + public FunctionFlowDefinition fromSupplier(Supplier messageSource) { + return fromSupplier(messageSource, null); + } + + public FunctionFlowDefinition fromSupplier(Supplier messageSource, + @Nullable Consumer endpointConfigurer) { + + return toFunctionFlow(IntegrationFlow.fromSupplier(messageSource, endpointConfigurer)); + } + + public FunctionFlowDefinition from(MessageChannel messageChannel) { + return toFunctionFlow(IntegrationFlow.from(messageChannel)); + } + + public FunctionFlowDefinition from(String messageChannelName) { + return from(messageChannelName, false); + } + + public FunctionFlowDefinition from(String messageChannelName, boolean fixedSubscriber) { + return toFunctionFlow(IntegrationFlow.from(messageChannelName, fixedSubscriber)); + } + + public FunctionFlowDefinition from(MessageSourceSpec> messageSourceSpec, + Consumer endpointConfigurer) { + + return toFunctionFlow(IntegrationFlow.from(messageSourceSpec, endpointConfigurer)); + } + + public FunctionFlowDefinition from(MessageSource messageSource) { + return from(messageSource, null); + } + + public FunctionFlowDefinition from(MessageSource messageSource, + @Nullable Consumer endpointConfigurer) { + + return toFunctionFlow(IntegrationFlow.from(messageSource, endpointConfigurer)); + } + + public FunctionFlowDefinition from(MessageProducerSupport messageProducer) { + return toFunctionFlow(IntegrationFlow.from(messageProducer)); + } + + public FunctionFlowDefinition from(MessagingGatewaySupport inboundGateway) { + return toFunctionFlow(IntegrationFlow.from(inboundGateway)); + } + + public FunctionFlowDefinition from(MessageChannelSpec messageChannelSpec) { + return toFunctionFlow(IntegrationFlow.from(messageChannelSpec)); + } + + public FunctionFlowDefinition from(MessageProducerSpec messageProducerSpec) { + return toFunctionFlow(IntegrationFlow.from(messageProducerSpec)); + } + + public FunctionFlowDefinition from(MessageSourceSpec> messageSourceSpec) { + return toFunctionFlow(IntegrationFlow.from(messageSourceSpec)); + } + + public FunctionFlowDefinition from(MessagingGatewaySpec inboundGatewaySpec) { + return toFunctionFlow(IntegrationFlow.from(inboundGatewaySpec)); + } + + public FunctionFlowDefinition from(Class serviceInterface) { + return from(serviceInterface, null); + } + + public FunctionFlowDefinition from(Class serviceInterface, + @Nullable Consumer endpointConfigurer) { + + return toFunctionFlow(IntegrationFlow.from(serviceInterface, endpointConfigurer)); + } + + public FunctionFlowDefinition from(Publisher> publisher) { + return toFunctionFlow(IntegrationFlow.from(publisher)); + } + + private FunctionFlowDefinition toFunctionFlow(IntegrationFlowBuilder from) { + FunctionFlowDefinition functionFlow = new FunctionFlowDefinition(this.functionLookupHelper); + from.channel(functionFlow.getInputChannel()); + functionFlow.addUpstreamComponents(from.get().getIntegrationComponents()); + return functionFlow; + } + +} diff --git a/spring-cloud-function-integration/src/main/java/org/springframework/cloud/function/integration/dsl/FunctionFlowDefinition.java b/spring-cloud-function-integration/src/main/java/org/springframework/cloud/function/integration/dsl/FunctionFlowDefinition.java new file mode 100644 index 000000000..cc8b84553 --- /dev/null +++ b/spring-cloud-function-integration/src/main/java/org/springframework/cloud/function/integration/dsl/FunctionFlowDefinition.java @@ -0,0 +1,94 @@ +/* + * Copyright 2023-2023 the original author or 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 + * + * https://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 org.springframework.cloud.function.integration.dsl; + +import java.util.Map; +import java.util.function.Consumer; +import java.util.function.Function; + +import org.springframework.integration.dsl.IntegrationFlow; +import org.springframework.integration.dsl.IntegrationFlowExtension; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; + +/** + * The {@link IntegrationFlowExtension} implementation for Spring Cloud Function domain. + * Adds operators for functions and consumers and overloaded versions based on their names + * or definitions resolved from the provided {@link org.springframework.cloud.function.context.FunctionCatalog}. + * + * @author Artem Bilan + * + * @since 4.0.3 + */ +public final class FunctionFlowDefinition extends IntegrationFlowExtension { + + private final FunctionLookupHelper functionLookupHelper; + + FunctionFlowDefinition(FunctionLookupHelper functionLookupHelper) { + this.functionLookupHelper = functionLookupHelper; + } + + MessageChannel getInputChannel() { + return getCurrentMessageChannel(); + } + + void addUpstreamComponents(Map components) { + addComponents(components); + } + + /** + * Configure a {@link org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry.FunctionInvocationWrapper} + * as a handler in the endpoint by its definition from the + * {@link org.springframework.cloud.function.context.FunctionCatalog}. + * @param functionDefinition the function definition in the function catalog. + * @return the current flow builder. + */ + public FunctionFlowDefinition apply(String functionDefinition) { + return apply(this.functionLookupHelper.lookupFunction(functionDefinition)); + } + + /** + * Configure a {@link Function} as a handler in the endpoint. + * @param function the {@link Function} to use. + * @return the current flow builder. + */ + public FunctionFlowDefinition apply(Function, ?> function) { + return handle(Message.class, (message, headers) -> function.apply(message)); + } + + /** + * Configure a {@link org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry.FunctionInvocationWrapper} + * as a one-way handler in the final endpoint by its definition from the + * {@link org.springframework.cloud.function.context.FunctionCatalog}. + * @param consumerDefinition the consumer definition in the function catalog. + * @return the current flow builder. + */ + public IntegrationFlow accept(String consumerDefinition) { + return accept(this.functionLookupHelper.lookupConsumer(consumerDefinition)); + } + + /** + * Configure a {@link Consumer} as a one-way handler in the final endpoint. + * @param consumer the {@link Consumer} to use. + * @return the current flow builder. + */ + public IntegrationFlow accept(Consumer> consumer) { + return handle(consumer::accept) + .get(); + } + +} diff --git a/spring-cloud-function-integration/src/main/java/org/springframework/cloud/function/integration/dsl/FunctionLookupHelper.java b/spring-cloud-function-integration/src/main/java/org/springframework/cloud/function/integration/dsl/FunctionLookupHelper.java new file mode 100644 index 000000000..60e965982 --- /dev/null +++ b/spring-cloud-function-integration/src/main/java/org/springframework/cloud/function/integration/dsl/FunctionLookupHelper.java @@ -0,0 +1,86 @@ +/* + * Copyright 2023-2023 the original author or 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 + * + * https://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 org.springframework.cloud.function.integration.dsl; + +import java.util.Objects; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Consumer; +import java.util.function.Function; +import java.util.function.Supplier; + +import org.springframework.cloud.function.context.FunctionCatalog; + +/** + * The helper class to lookup functions from the catalog in lazy manner and cache their instances. + * + * @author Artem Bilan + * + * @since 4.0.3 + */ +public class FunctionLookupHelper { + + private final FunctionCatalog functionCatalog; + + FunctionLookupHelper(FunctionCatalog functionCatalog) { + this.functionCatalog = functionCatalog; + } + +

Supplier

lookupSupplier(String functionDefinition) { + return () -> + memoize(() -> this.functionCatalog.>lookup(Supplier.class, functionDefinition)) + .get() + .get(); + } + +

Function lookupFunction(String functionDefinition) { + return (p) -> + memoize(() -> this.functionCatalog.>lookup(Function.class, functionDefinition)) + .get() + .apply(p); + } + +

Consumer

lookupConsumer(String consumerDefinition) { + return (p) -> + memoize(() -> this.functionCatalog.>lookup(Consumer.class, consumerDefinition)) + .get() + .accept(p); + } + + /** + * The delegate {@link Supplier#get()} is called exactly once and the result is cached. + * @param Generic type of supplied value + * @param delegate The actual Supplier + * @return The memoized Supplier + */ + private static Supplier memoize(Supplier delegate) { + AtomicReference value = new AtomicReference<>(); + return () -> { + T val = value.get(); + if (val == null) { + synchronized (value) { + val = value.get(); + if (val == null) { + val = Objects.requireNonNull(delegate.get()); + value.set(val); + } + } + } + return val; + }; + } + +} diff --git a/spring-cloud-function-integration/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports b/spring-cloud-function-integration/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports new file mode 100644 index 000000000..b7d39cea2 --- /dev/null +++ b/spring-cloud-function-integration/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports @@ -0,0 +1 @@ +org.springframework.cloud.function.integration.dsl.FunctionFlowAutoConfiguration diff --git a/spring-cloud-function-integration/src/test/java/org/springframework/cloud/function/integration/dsl/FunctionFlowTests.java b/spring-cloud-function-integration/src/test/java/org/springframework/cloud/function/integration/dsl/FunctionFlowTests.java new file mode 100644 index 000000000..861c7510c --- /dev/null +++ b/spring-cloud-function-integration/src/test/java/org/springframework/cloud/function/integration/dsl/FunctionFlowTests.java @@ -0,0 +1,136 @@ +/* + * Copyright 2023-2023 the original author or 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 + * + * https://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 org.springframework.cloud.function.integration.dsl; + +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; +import java.util.function.Consumer; +import java.util.function.Function; +import java.util.function.Supplier; + +import org.junit.jupiter.api.Test; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.dsl.IntegrationFlow; +import org.springframework.integration.dsl.PollerSpec; +import org.springframework.integration.dsl.Pollers; +import org.springframework.integration.handler.LoggingHandler; +import org.springframework.integration.scheduling.PollerMetadata; +import org.springframework.integration.test.util.OnlyOnceTrigger; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.support.GenericMessage; +import org.springframework.test.annotation.DirtiesContext; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * @author Artem Bilan + * + * @since 4.0.3 + */ +@SpringBootTest +@DirtiesContext +public class FunctionFlowTests { + + @Autowired + QueueChannel wireTapChannel; + + @Autowired + BlockingQueue results; + + @Test + void fromSupplierOverFunctionToConsumer() throws InterruptedException { + String result = results.poll(10, TimeUnit.SECONDS); + assertThat(result).isEqualTo("SIMPLE TEST DATA"); + Message receive = wireTapChannel.receive(10_000); + assertThat(receive).isNotNull() + .extracting(Message::getPayload) + .isEqualTo("simple test data".getBytes()); + } + + @Autowired + MessageChannel functionCompositionInput; + + @Test + void fromChannelToFunctionComposition() throws InterruptedException { + this.functionCompositionInput.send(new GenericMessage<>("compose this")); + + String result = results.poll(10, TimeUnit.SECONDS); + assertThat(result).isEqualTo("COMPOSE THIS"); + } + + @EnableAutoConfiguration + @Configuration(proxyBeanMethods = false) + static class TestIntegrationConfiguration { + + @Bean(PollerMetadata.DEFAULT_POLLER) + PollerSpec defaultPoller() { + return Pollers.trigger(new OnlyOnceTrigger()); + } + + @Bean + Supplier simpleByteArraySupplier() { + return "simple test data"::getBytes; + } + + @Bean + Function upperCaseFunction() { + return String::toUpperCase; + } + + @Bean + BlockingQueue results() { + return new LinkedBlockingQueue<>(); + } + + @Bean + Consumer simpleStringConsumer(BlockingQueue results) { + return results::add; + } + + @Bean + QueueChannel wireTapChannel() { + return new QueueChannel(); + } + + @Bean + IntegrationFlow someFunctionFlow(FunctionFlowBuilder functionFlowBuilder) { + return functionFlowBuilder + .fromSupplier("simpleByteArraySupplier") + .wireTap("wireTapChannel") + .apply("upperCaseFunction") + .log(LoggingHandler.Level.WARN, FunctionFlowTests.class.getName()) + .accept("simpleStringConsumer"); + } + + @Bean + IntegrationFlow functionCompositionFlow(FunctionFlowBuilder functionFlowBuilder) { + return functionFlowBuilder + .from("functionCompositionInput") + .accept("upperCaseFunction|simpleStringConsumer"); + } + + } + +}