GH-1440: Add support for Supplier as Source
Resolves spring-cloud/spring-cloud-stream#1440 Added initial support and configuration to create Source apps by simply providing a Supplier bean while also benefiting from composition provided by SCF Refactored/renamed some of the new classes/methods addressed PR comments * Polishing code style and fix typos
This commit is contained in:
Oleg Zhurakousky
committed by
Artem Bilan
Artem Bilan
parent
dcd5259041
commit
6b1d584c6c
@@ -34,10 +34,12 @@ import org.springframework.context.annotation.Configuration;
|
||||
public class FunctionConfiguration {
|
||||
|
||||
@Bean
|
||||
public FunctionSupport functionSupport(FunctionCatalog functionCatalog, FunctionInspector functionInspector,
|
||||
CompositeMessageConverterFactory messageConverterFactory) {
|
||||
public IntegrationFlowFunctionSupport functionSupport(FunctionCatalog functionCatalog,
|
||||
FunctionInspector functionInspector, CompositeMessageConverterFactory messageConverterFactory,
|
||||
FunctionProperties functionProperties) {
|
||||
|
||||
return new FunctionSupport(functionCatalog, functionInspector, messageConverterFactory);
|
||||
return new IntegrationFlowFunctionSupport(functionCatalog, functionInspector, messageConverterFactory,
|
||||
functionProperties);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -1,86 +0,0 @@
|
||||
/*
|
||||
* Copyright 2018 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
|
||||
*
|
||||
* 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 org.springframework.cloud.stream.function;
|
||||
|
||||
import java.util.function.Consumer;
|
||||
|
||||
import org.reactivestreams.Publisher;
|
||||
|
||||
import reactor.core.publisher.Flux;
|
||||
import reactor.core.publisher.Mono;
|
||||
|
||||
import org.springframework.cloud.function.context.FunctionCatalog;
|
||||
import org.springframework.cloud.function.context.catalog.FunctionInspector;
|
||||
import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory;
|
||||
import org.springframework.integration.dsl.IntegrationFlowBuilder;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.util.Assert;
|
||||
|
||||
|
||||
|
||||
/**
|
||||
*
|
||||
* @author Oleg Zhurakousky
|
||||
*
|
||||
* @since 2.1
|
||||
*/
|
||||
public class FunctionSupport {
|
||||
|
||||
private final FunctionCatalog functionCatalog;
|
||||
|
||||
private final FunctionInspector functionInspector;
|
||||
|
||||
private final CompositeMessageConverterFactory messageConverterFactory;
|
||||
|
||||
public FunctionSupport(FunctionCatalog functionCatalog, FunctionInspector functionInspector,
|
||||
CompositeMessageConverterFactory messageConverterFactory) {
|
||||
|
||||
Assert.notNull(functionCatalog, "'functionCatalog' must not be null");
|
||||
Assert.notNull(functionInspector, "'functionInspector' must not be null");
|
||||
Assert.notNull(messageConverterFactory, "'messageConverterFactory' must not be null");
|
||||
this.functionCatalog = functionCatalog;
|
||||
this.functionInspector = functionInspector;
|
||||
this.messageConverterFactory = messageConverterFactory;
|
||||
}
|
||||
|
||||
public <I,O> void applyFunctionToIntegrationFlow(String functionName, IntegrationFlowBuilder flowBuilder,
|
||||
Consumer<Message<O>> outputProcessor) {
|
||||
|
||||
FunctionInvoker<I,O> functionInvoker =
|
||||
new FunctionInvoker<>(functionName, this.functionCatalog, this.functionInspector,
|
||||
this.messageConverterFactory);
|
||||
|
||||
subscribeToInput(functionInvoker, flowBuilder.toReactivePublisher(), outputProcessor);
|
||||
}
|
||||
|
||||
private <O> Mono<Void> subscribeToOutput(Consumer<Message<O>> outputProcessor,
|
||||
Publisher<Message<O>> outputPublisher) {
|
||||
|
||||
Flux<Message<O>> output = outputProcessor == null
|
||||
? Flux.from(outputPublisher)
|
||||
: Flux.from(outputPublisher).doOnNext(outputProcessor);
|
||||
return output.then();
|
||||
}
|
||||
|
||||
private <I,O> void subscribeToInput(FunctionInvoker<I,O> functionInvoker, Publisher<Message<I>> publisher,
|
||||
Consumer<Message<O>> outputProcessor) {
|
||||
|
||||
Flux<Message<I>> inputPublisher = Flux.from(publisher);
|
||||
subscribeToOutput(outputProcessor, functionInvoker.apply(inputPublisher)).subscribe();
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,142 @@
|
||||
/*
|
||||
* Copyright 2018 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
|
||||
*
|
||||
* 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 org.springframework.cloud.stream.function;
|
||||
|
||||
import java.util.function.Consumer;
|
||||
import java.util.function.Function;
|
||||
import java.util.function.Supplier;
|
||||
|
||||
import org.reactivestreams.Publisher;
|
||||
|
||||
import reactor.core.publisher.Flux;
|
||||
import reactor.core.publisher.Mono;
|
||||
|
||||
import org.springframework.cloud.function.context.FunctionCatalog;
|
||||
import org.springframework.cloud.function.context.catalog.FunctionInspector;
|
||||
import org.springframework.cloud.function.core.FluxSupplier;
|
||||
import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory;
|
||||
import org.springframework.integration.dsl.IntegrationFlowBuilder;
|
||||
import org.springframework.integration.dsl.IntegrationFlows;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.MessageChannel;
|
||||
import org.springframework.util.Assert;
|
||||
import org.springframework.util.StringUtils;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author Oleg Zhurakousky
|
||||
*
|
||||
* @since 2.1
|
||||
*/
|
||||
public class IntegrationFlowFunctionSupport {
|
||||
|
||||
private final FunctionCatalog functionCatalog;
|
||||
|
||||
private final FunctionInspector functionInspector;
|
||||
|
||||
private final CompositeMessageConverterFactory messageConverterFactory;
|
||||
|
||||
private final FunctionProperties functionProperties;
|
||||
|
||||
/**
|
||||
* @param functionCatalog
|
||||
* @param functionInspector
|
||||
* @param messageConverterFactory
|
||||
* @param functionProperties
|
||||
*/
|
||||
public IntegrationFlowFunctionSupport(FunctionCatalog functionCatalog, FunctionInspector functionInspector,
|
||||
CompositeMessageConverterFactory messageConverterFactory, FunctionProperties functionProperties) {
|
||||
|
||||
Assert.notNull(functionCatalog, "'functionCatalog' must not be null");
|
||||
Assert.notNull(functionInspector, "'functionInspector' must not be null");
|
||||
Assert.notNull(messageConverterFactory, "'messageConverterFactory' must not be null");
|
||||
Assert.notNull(functionProperties, "'functionProperties' must not be null");
|
||||
this.functionCatalog = functionCatalog;
|
||||
this.functionInspector = functionInspector;
|
||||
this.messageConverterFactory = messageConverterFactory;
|
||||
this.functionProperties = functionProperties;
|
||||
}
|
||||
|
||||
/**
|
||||
* Create an instance of the {@link IntegrationFlowBuilder} from a {@link Supplier} bean available in the context.
|
||||
* The name of the bean must be provided via `spring.cloud.stream.function.name` property.
|
||||
* @return instance of {@link IntegrationFlowBuilder}
|
||||
* @throws IllegalStateException if the named Supplier can not be located.
|
||||
*/
|
||||
public IntegrationFlowBuilder integrationFlowFromNamedSupplier() {
|
||||
if (StringUtils.hasText(this.functionProperties.getName())) {
|
||||
Supplier<?> supplier = functionCatalog.lookup(Supplier.class, this.functionProperties.getName());
|
||||
if (supplier instanceof FluxSupplier) {
|
||||
supplier = ((FluxSupplier<?>)supplier).getTarget();
|
||||
}
|
||||
return integrationFlowFromProvidedSupplier(supplier)
|
||||
.split();
|
||||
}
|
||||
|
||||
throw new IllegalStateException(
|
||||
"A Supplier is not specified in the 'spring.cloud.stream.function.name' property.");
|
||||
}
|
||||
|
||||
/**
|
||||
* Create an instance of the {@link IntegrationFlowBuilder} from a provided {@link Supplier}.
|
||||
* @return instance of {@link IntegrationFlowBuilder}
|
||||
*/
|
||||
public IntegrationFlowBuilder integrationFlowFromProvidedSupplier(Supplier<?> supplier) {
|
||||
return IntegrationFlows.from(supplier);
|
||||
}
|
||||
|
||||
/**
|
||||
* Add a {@link Function} bean to the end of an integration flow.
|
||||
* The name of the bean must be provided via `spring.cloud.stream.function.name` property.
|
||||
* <p>
|
||||
* NOTE: If this method returns true, the integration flow is now represented
|
||||
* as a Reactive Streams {@link Publisher} bean.
|
||||
* </p>
|
||||
* @param flowBuilder instance of the {@link IntegrationFlowBuilder} representing
|
||||
* the current state of the integration flow
|
||||
* @param outputChannel channel where the output of a function will be sent
|
||||
* @return true if {@link Function} was located and added and false if it wasn't.
|
||||
*/
|
||||
public <I,O> boolean andThenFunction(IntegrationFlowBuilder flowBuilder, MessageChannel outputChannel) {
|
||||
if (StringUtils.hasText(this.functionProperties.getName())) {
|
||||
FunctionInvoker<I,O> functionInvoker =
|
||||
new FunctionInvoker<>(this.functionProperties.getName(), this.functionCatalog,
|
||||
this.functionInspector, this.messageConverterFactory);
|
||||
|
||||
subscribeToInput(functionInvoker, flowBuilder.toReactivePublisher(), outputChannel::send);
|
||||
return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
private <O> Mono<Void> subscribeToOutput(Consumer<Message<O>> outputProcessor,
|
||||
Publisher<Message<O>> outputPublisher) {
|
||||
|
||||
Flux<Message<O>> output = outputProcessor == null
|
||||
? Flux.from(outputPublisher)
|
||||
: Flux.from(outputPublisher).doOnNext(outputProcessor);
|
||||
return output.then();
|
||||
}
|
||||
|
||||
private <I,O> void subscribeToInput(FunctionInvoker<I,O> functionInvoker, Publisher<Message<I>> publisher,
|
||||
Consumer<Message<O>> outputProcessor) {
|
||||
|
||||
Flux<Message<I>> inputPublisher = Flux.from(publisher);
|
||||
subscribeToOutput(outputProcessor, functionInvoker.apply(inputPublisher)).subscribe();
|
||||
}
|
||||
|
||||
}
|
||||
@@ -1,134 +0,0 @@
|
||||
/*
|
||||
* Copyright 2018 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
|
||||
*
|
||||
* 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 org.springframework.cloud.stream.function;
|
||||
|
||||
import java.nio.charset.StandardCharsets;
|
||||
|
||||
import java.util.function.Function;
|
||||
import java.util.function.Supplier;
|
||||
|
||||
import org.junit.Test;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.boot.WebApplicationType;
|
||||
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||
import org.springframework.boot.builder.SpringApplicationBuilder;
|
||||
import org.springframework.cloud.stream.annotation.EnableBinding;
|
||||
import org.springframework.cloud.stream.binder.test.OutputDestination;
|
||||
import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
|
||||
import org.springframework.cloud.stream.messaging.Source;
|
||||
import org.springframework.context.ApplicationContext;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Import;
|
||||
import org.springframework.integration.dsl.IntegrationFlow;
|
||||
import org.springframework.integration.dsl.IntegrationFlowBuilder;
|
||||
import org.springframework.integration.dsl.IntegrationFlows;
|
||||
import org.springframework.integration.support.MessageBuilder;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.MessageHeaders;
|
||||
import org.springframework.util.MimeTypeUtils;
|
||||
import org.springframework.util.StringUtils;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author Oleg Zhurakousky
|
||||
*
|
||||
* @since 2.1
|
||||
*
|
||||
*/
|
||||
public class FunctionSupportTests {
|
||||
|
||||
@Test
|
||||
public void testFunctionIsAppliedToMessageSource() {
|
||||
ApplicationContext context =
|
||||
new SpringApplicationBuilder(
|
||||
TestChannelBinderConfiguration.getCompleteConfiguration(MyFunctionsConfiguration.class))
|
||||
.web(WebApplicationType.NONE)
|
||||
.run("--spring.cloud.stream.function.name=toUpperCase", "--spring.jmx.enabled=false");
|
||||
|
||||
OutputDestination target = context.getBean(OutputDestination.class);
|
||||
assertThat(target.receive(1000).getPayload()).isEqualTo("HELLO FUNCTION".getBytes(StandardCharsets.UTF_8));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testComposedFunctionIsAppliedToMessageSource() {
|
||||
ApplicationContext context =
|
||||
new SpringApplicationBuilder(
|
||||
TestChannelBinderConfiguration.getCompleteConfiguration(MyFunctionsConfiguration.class))
|
||||
.web(WebApplicationType.NONE)
|
||||
.run("--spring.cloud.stream.function.name=toUpperCase|concatWithSelf",
|
||||
"--spring.jmx.enabled=false");
|
||||
|
||||
OutputDestination target = context.getBean(OutputDestination.class);
|
||||
assertThat(target.receive(1000).getPayload())
|
||||
.isEqualTo("HELLO FUNCTION:HELLO FUNCTION".getBytes(StandardCharsets.UTF_8));
|
||||
}
|
||||
|
||||
@SpringBootApplication
|
||||
@Import(SourceConfiguration.class)
|
||||
public static class MyFunctionsConfiguration {
|
||||
|
||||
@Bean
|
||||
public Function<String, String> toUpperCase() {
|
||||
return String::toUpperCase;
|
||||
}
|
||||
|
||||
@Bean
|
||||
public Function<String, String> concatWithSelf() {
|
||||
return x -> x + ":" + x;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
/**
|
||||
* This configuration essentially emulates our existing app-starters for Sources
|
||||
* and essentially demonstrates how a function(s) could be applied to an existing
|
||||
* source via {@link FunctionSupport} class.
|
||||
*/
|
||||
@EnableBinding(Source.class)
|
||||
public static class SourceConfiguration {
|
||||
|
||||
@Autowired
|
||||
private Source source;
|
||||
|
||||
@Autowired
|
||||
private FunctionProperties functionProperties;
|
||||
|
||||
@Bean
|
||||
public IntegrationFlow messageSourceFlow(FunctionSupport functionSupport) {
|
||||
Supplier<Message<String>> messageSource = () -> MessageBuilder.withPayload("hello function")
|
||||
.setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN).build();
|
||||
|
||||
IntegrationFlowBuilder flowBuilder = IntegrationFlows.from(messageSource);
|
||||
|
||||
if (StringUtils.hasText(functionProperties.getName())) {
|
||||
functionSupport
|
||||
.applyFunctionToIntegrationFlow(functionProperties.getName(), flowBuilder,
|
||||
message -> source.output().send(message));
|
||||
}
|
||||
else {
|
||||
flowBuilder = flowBuilder.channel(source.output());
|
||||
}
|
||||
|
||||
return flowBuilder.get();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,232 @@
|
||||
/*
|
||||
* Copyright 2018 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
|
||||
*
|
||||
* 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 org.springframework.cloud.stream.function;
|
||||
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
import java.util.function.Function;
|
||||
import java.util.function.Supplier;
|
||||
|
||||
import org.junit.Test;
|
||||
|
||||
import reactor.core.publisher.Flux;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.boot.WebApplicationType;
|
||||
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
|
||||
import org.springframework.boot.builder.SpringApplicationBuilder;
|
||||
import org.springframework.cloud.stream.annotation.EnableBinding;
|
||||
import org.springframework.cloud.stream.binder.test.OutputDestination;
|
||||
import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
|
||||
import org.springframework.cloud.stream.messaging.Source;
|
||||
import org.springframework.context.ConfigurableApplicationContext;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Import;
|
||||
import org.springframework.integration.dsl.IntegrationFlow;
|
||||
import org.springframework.integration.dsl.IntegrationFlowBuilder;
|
||||
import org.springframework.integration.support.MessageBuilder;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.MessageHeaders;
|
||||
import org.springframework.util.Assert;
|
||||
import org.springframework.util.MimeTypeUtils;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author Oleg Zhurakousky
|
||||
*
|
||||
* @since 2.1
|
||||
*
|
||||
*/
|
||||
public class SourceToFunctionsSupportTests {
|
||||
|
||||
@Test
|
||||
public void testFunctionIsAppliedToExistingMessageSource() {
|
||||
try (ConfigurableApplicationContext context =
|
||||
new SpringApplicationBuilder(
|
||||
TestChannelBinderConfiguration.getCompleteConfiguration(FunctionsConfiguration.class))
|
||||
.web(WebApplicationType.NONE)
|
||||
.run("--spring.cloud.stream.function.name=toUpperCase", "--spring.jmx.enabled=false")) {
|
||||
|
||||
OutputDestination target = context.getBean(OutputDestination.class);
|
||||
assertThat(target.receive(1000).getPayload()).isEqualTo("HELLO FUNCTION".getBytes(StandardCharsets.UTF_8));
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testComposedFunctionIsAppliedToExistingMessageSource() {
|
||||
try (ConfigurableApplicationContext context =
|
||||
new SpringApplicationBuilder(
|
||||
TestChannelBinderConfiguration.getCompleteConfiguration(FunctionsConfiguration.class))
|
||||
.web(WebApplicationType.NONE)
|
||||
.run("--spring.cloud.stream.function.name=toUpperCase|concatWithSelf",
|
||||
"--spring.jmx.enabled=false")) {
|
||||
OutputDestination target = context.getBean(OutputDestination.class);
|
||||
assertThat(target.receive(1000).getPayload())
|
||||
.isEqualTo("HELLO FUNCTION:HELLO FUNCTION".getBytes(StandardCharsets.UTF_8));
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testMessageSourceIsCreatedFromProvidedSupplier() {
|
||||
try (ConfigurableApplicationContext context =
|
||||
new SpringApplicationBuilder(
|
||||
TestChannelBinderConfiguration.getCompleteConfiguration(SupplierConfiguration.class))
|
||||
.web(WebApplicationType.NONE)
|
||||
.run("--spring.cloud.stream.function.name=number",
|
||||
"--spring.jmx.enabled=false")) {
|
||||
|
||||
OutputDestination target = context.getBean(OutputDestination.class);
|
||||
assertThat(target.receive(10000).getPayload())
|
||||
.isEqualTo("1".getBytes(StandardCharsets.UTF_8));
|
||||
assertThat(target.receive(10000).getPayload())
|
||||
.isEqualTo("2".getBytes(StandardCharsets.UTF_8));
|
||||
assertThat(target.receive(10000).getPayload())
|
||||
.isEqualTo("3".getBytes(StandardCharsets.UTF_8));
|
||||
//etc
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testMessageSourceIsCreatedFromProvidedSupplierComposedWithSingleFunction() {
|
||||
try (ConfigurableApplicationContext context =
|
||||
new SpringApplicationBuilder(
|
||||
TestChannelBinderConfiguration.getCompleteConfiguration(SupplierConfiguration.class))
|
||||
.web(WebApplicationType.NONE)
|
||||
.run("--spring.cloud.stream.function.name=number|concatWithSelf",
|
||||
"--spring.jmx.enabled=false")) {
|
||||
|
||||
OutputDestination target = context.getBean(OutputDestination.class);
|
||||
assertThat(target.receive(10000).getPayload())
|
||||
.isEqualTo("11".getBytes(StandardCharsets.UTF_8));
|
||||
assertThat(target.receive(10000).getPayload())
|
||||
.isEqualTo("22".getBytes(StandardCharsets.UTF_8));
|
||||
assertThat(target.receive(10000).getPayload())
|
||||
.isEqualTo("33".getBytes(StandardCharsets.UTF_8));
|
||||
//etc
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testMessageSourceIsCreatedFromProvidedSupplierComposedWithMultipleFunctions() {
|
||||
try (ConfigurableApplicationContext context =
|
||||
new SpringApplicationBuilder(
|
||||
TestChannelBinderConfiguration.getCompleteConfiguration(SupplierConfiguration.class))
|
||||
.web(WebApplicationType.NONE)
|
||||
.run("--spring.cloud.stream.function.name=number|concatWithSelf|multiplyByTwo",
|
||||
"--spring.jmx.enabled=false")) {
|
||||
|
||||
OutputDestination target = context.getBean(OutputDestination.class);
|
||||
assertThat(target.receive(10000).getPayload())
|
||||
.isEqualTo("22".getBytes(StandardCharsets.UTF_8));
|
||||
assertThat(target.receive(10000).getPayload())
|
||||
.isEqualTo("44".getBytes(StandardCharsets.UTF_8));
|
||||
assertThat(target.receive(10000).getPayload())
|
||||
.isEqualTo("66".getBytes(StandardCharsets.UTF_8));
|
||||
//etc
|
||||
}
|
||||
}
|
||||
|
||||
@EnableAutoConfiguration
|
||||
@Import(ProvidedMessageSourceConfiguration.class)
|
||||
public static class SupplierConfiguration {
|
||||
|
||||
AtomicInteger counter = new AtomicInteger();
|
||||
|
||||
@Bean
|
||||
public Supplier<String> number() {
|
||||
return () -> String.valueOf(counter.incrementAndGet());
|
||||
}
|
||||
|
||||
@Bean
|
||||
public Function<String, String> concatWithSelf() {
|
||||
return x -> x + x;
|
||||
}
|
||||
|
||||
@Bean
|
||||
public Function<Flux<String>, Flux<String>> multiplyByTwo() {
|
||||
return x -> x.map(i -> String.valueOf(Integer.valueOf(i) * 2));
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@EnableAutoConfiguration
|
||||
@Import(ExistingMessageSourceConfiguration.class)
|
||||
public static class FunctionsConfiguration {
|
||||
|
||||
@Bean
|
||||
public Function<String, String> toUpperCase() {
|
||||
return String::toUpperCase;
|
||||
}
|
||||
|
||||
@Bean
|
||||
public Function<String, String> concatWithSelf() {
|
||||
return x -> x + ":" + x;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
/**
|
||||
* This configuration essentially emulates our existing app-starters for Sources
|
||||
* and essentially demonstrates how a function(s) could be applied to an existing
|
||||
* source via {@link IntegrationFlowFunctionSupport} class.
|
||||
*/
|
||||
@EnableBinding(Source.class)
|
||||
public static class ExistingMessageSourceConfiguration {
|
||||
|
||||
@Autowired
|
||||
private Source source;
|
||||
|
||||
@Bean
|
||||
public IntegrationFlow messageSourceFlow(IntegrationFlowFunctionSupport functionSupport) {
|
||||
Supplier<Message<String>> messageSource = () -> MessageBuilder.withPayload("hello function")
|
||||
.setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN).build();
|
||||
|
||||
IntegrationFlowBuilder flowBuilder = functionSupport.integrationFlowFromProvidedSupplier(messageSource);
|
||||
|
||||
if (!functionSupport.andThenFunction(flowBuilder, this.source.output())) {
|
||||
flowBuilder = flowBuilder.channel(this.source.output());
|
||||
}
|
||||
|
||||
return flowBuilder.get();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@EnableBinding(Source.class)
|
||||
public static class ProvidedMessageSourceConfiguration {
|
||||
|
||||
@Autowired
|
||||
private Source source;
|
||||
|
||||
@Autowired
|
||||
private FunctionProperties functionProperties;
|
||||
|
||||
@Bean
|
||||
public IntegrationFlow messageSourceFlow(IntegrationFlowFunctionSupport functionSupport) {
|
||||
Assert.hasText(this.functionProperties.getName(), "Supplier name must be provided");
|
||||
|
||||
return functionSupport
|
||||
.integrationFlowFromNamedSupplier()
|
||||
.channel(this.source.output())
|
||||
.get();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user