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

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);
+	}
+
+}

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.
+ * <p>
+ * 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}.
+ * <p>
+ * 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:
+ * <pre class="code">
+ * {@code
+ * @Bean
+ * IntegrationFlow someFunctionFlow(FunctionFlowBuilder functionFlowBuilder) {
+ *		return functionFlowBuilder
+ *				.fromSupplier("timeSupplier")
+ *				.apply("spelFunction")
+ *				.log(LoggingHandler.Level.DEBUG, "some.log.category")
+ *				.<String, String>transform(String::toUpperCase)
+ *				.accept("fileConsumer");
+ * }
+ * }
+ * </pre>
+ *
+ * @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<SourcePollingChannelAdapterSpec> endpointConfigurer) {
+
+		return fromSupplier(this.functionLookupHelper.lookupSupplier(supplierDefinition), endpointConfigurer);
+	}
+
+	public <T> FunctionFlowDefinition fromSupplier(Supplier<T> messageSource) {
+		return fromSupplier(messageSource, null);
+	}
+
+	public <T> FunctionFlowDefinition fromSupplier(Supplier<T> messageSource,
+			@Nullable Consumer<SourcePollingChannelAdapterSpec> 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<?, ? extends MessageSource<?>> messageSourceSpec,
+			Consumer<SourcePollingChannelAdapterSpec> endpointConfigurer) {
+
+		return toFunctionFlow(IntegrationFlow.from(messageSourceSpec, endpointConfigurer));
+	}
+
+	public FunctionFlowDefinition from(MessageSource<?> messageSource) {
+		return from(messageSource, null);
+	}
+
+	public FunctionFlowDefinition from(MessageSource<?> messageSource,
+			@Nullable Consumer<SourcePollingChannelAdapterSpec> 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<?, ? extends MessageSource<?>> 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<GatewayProxySpec> endpointConfigurer) {
+
+		return toFunctionFlow(IntegrationFlow.from(serviceInterface, endpointConfigurer));
+	}
+
+	public FunctionFlowDefinition from(Publisher<? extends Message<?>> 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;
+	}
+
+}

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<FunctionFlowDefinition> {
+
+	private final FunctionLookupHelper functionLookupHelper;
+
+	FunctionFlowDefinition(FunctionLookupHelper functionLookupHelper) {
+		this.functionLookupHelper = functionLookupHelper;
+	}
+
+	MessageChannel getInputChannel() {
+		return getCurrentMessageChannel();
+	}
+
+	void addUpstreamComponents(Map<Object, String> 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<Message<?>, ?> 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<Message<?>> consumer) {
+		return handle(consumer::accept)
+				.get();
+	}
+
+}

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;
+	}
+
+	<P> Supplier<P> lookupSupplier(String functionDefinition) {
+		return () ->
+				memoize(() -> this.functionCatalog.<Supplier<P>>lookup(Supplier.class, functionDefinition))
+						.get()
+						.get();
+	}
+
+	<P> Function<P, ?> lookupFunction(String functionDefinition) {
+		return (p) ->
+				memoize(() -> this.functionCatalog.<Function<P, ?>>lookup(Function.class, functionDefinition))
+						.get()
+						.apply(p);
+	}
+
+	<P> Consumer<P> lookupConsumer(String consumerDefinition) {
+		return (p) ->
+				memoize(() -> this.functionCatalog.<Consumer<P>>lookup(Consumer.class, consumerDefinition))
+						.get()
+						.accept(p);
+	}
+
+	/**
+	 * The delegate {@link Supplier#get()} is called exactly once and the result is cached.
+	 * @param  <T> Generic type of supplied value
+	 * @param  delegate The actual Supplier
+	 * @return The memoized Supplier
+	 */
+	private static <T> Supplier<T> memoize(Supplier<? extends T> delegate) {
+		AtomicReference<T> 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;
+		};
+	}
+
+}

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

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<String> 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<byte[]> simpleByteArraySupplier() {
+			return "simple test data"::getBytes;
+		}
+
+		@Bean
+		Function<String, String> upperCaseFunction() {
+			return String::toUpperCase;
+		}
+
+		@Bean
+		BlockingQueue<String> results() {
+			return new LinkedBlockingQueue<>();
+		}
+
+		@Bean
+		Consumer<String> simpleStringConsumer(BlockingQueue<String> 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");
+		}
+
+	}
+
+}