From c8a4b40de0026f5aff0da341b8c9c31af88e7560 Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Mon, 29 Jul 2019 15:00:15 +0200 Subject: [PATCH] Fixed source and source-to-function functionality and tests --- .../function/FunctionConfiguration.java | 111 +++++-- .../IntegrationFlowFunctionSupport.java | 3 +- .../SourceToFunctionsSupportTests.java | 311 +++++++----------- 3 files changed, 206 insertions(+), 219 deletions(-) diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java index 4f22a159a..7d5a6a030 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java @@ -18,14 +18,17 @@ package org.springframework.cloud.stream.function; import java.lang.reflect.Type; import java.time.Duration; +import java.util.concurrent.atomic.AtomicReference; import java.util.function.Consumer; import java.util.function.Function; import java.util.function.Supplier; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; import org.reactivestreams.Publisher; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; - +import reactor.core.publisher.MonoSink; import org.springframework.beans.BeansException; import org.springframework.beans.factory.config.BeanPostProcessor; @@ -35,6 +38,7 @@ import org.springframework.cloud.function.context.FunctionCatalog; import org.springframework.cloud.function.context.catalog.BeanFactoryAwareFunctionRegistry.FunctionInvocationWrapper; import org.springframework.cloud.function.context.catalog.FunctionInspector; import org.springframework.cloud.function.context.catalog.FunctionTypeUtils; +import org.springframework.cloud.stream.binder.BindingCreatedEvent; import org.springframework.cloud.stream.binding.BindableProxyFactory; import org.springframework.cloud.stream.config.BinderFactoryAutoConfiguration; import org.springframework.cloud.stream.config.BindingServiceConfiguration; @@ -46,17 +50,20 @@ import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Import; import org.springframework.context.support.GenericApplicationContext; import org.springframework.integration.channel.MessageChannelReactiveUtils; +import org.springframework.integration.dsl.IntegrationFlow; +import org.springframework.integration.dsl.IntegrationFlowBuilder; +import org.springframework.integration.dsl.IntegrationFlows; import org.springframework.integration.handler.ServiceActivatingHandler; +import org.springframework.integration.support.MessageBuilder; import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.SubscribableChannel; import org.springframework.util.Assert; +import org.springframework.util.MimeTypeUtils; import org.springframework.util.ObjectUtils; - - - /** * @author Oleg Zhurakousky * @author David Turanski @@ -76,8 +83,68 @@ public class FunctionConfiguration { ObjectUtils.isEmpty(bindableProxyFactory) ? null : bindableProxyFactory[0]); } + @Bean + public IntegrationFlow standAloneSupplierFlow(FunctionCatalog functionCatalog, FunctionInspector functionInspector, + StreamFunctionProperties functionProperties, GenericApplicationContext context) { + IntegrationFlow integrationFlow = null; + FunctionInvocationWrapper functionWrapper = functionCatalog.lookup(functionProperties.getDefinition()); + if (functionWrapper != null) { + AtomicReference> triggerRef = new AtomicReference<>(); + Publisher beginPublishingTrigger = Mono.create(emmiter -> { + triggerRef.set(emmiter); + }); + context.addApplicationListener(event -> { + if (event instanceof BindingCreatedEvent) { + if (triggerRef.get() != null) { + triggerRef.get().success(); + } + } + }); + + if (!functionProperties.isComposeFrom() && !functionProperties.isComposeTo() && functionWrapper.isSupplier()) { + integrationFlow = this.integrationFlowFromProvidedSupplier(functionWrapper, functionInspector, beginPublishingTrigger) + .channel("output").get(); + } + } + + return integrationFlow; + } + + @SuppressWarnings({ "rawtypes", "unchecked" }) + private IntegrationFlowBuilder integrationFlowFromProvidedSupplier(Supplier supplier, + FunctionInspector inspector, Publisher beginPublishingTrigger) { + + IntegrationFlowBuilder integrationFlowBuilder; + Type functionType = FunctionTypeUtils.getFunctionType(supplier, inspector); + if (FunctionTypeUtils.isReactive(FunctionTypeUtils.getInputType(functionType, 0))) { + Publisher publisher = (Publisher) supplier.get(); + publisher = publisher instanceof Mono + ? ((Mono) publisher).delaySubscription(beginPublishingTrigger).map(this::wrapToMessageIfNecessary) + : ((Flux) publisher).delaySubscription(beginPublishingTrigger).map(this::wrapToMessageIfNecessary); + + integrationFlowBuilder = IntegrationFlows.from(publisher); + } + else { + integrationFlowBuilder = IntegrationFlows.from(supplier); + } + return integrationFlowBuilder; + } + + @SuppressWarnings("unchecked") + private Message wrapToMessageIfNecessary(T value) { + return value instanceof Message ? (Message) value : MessageBuilder.withPayload(value).setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON).build(); + } + + + /** + * + * @author Oleg Zhurakousky + * @since 3.0 + */ private static class FunctionChannelBindingPostProcessor implements BeanPostProcessor, ApplicationContextAware { + private static Log logger = LogFactory.getLog(FunctionChannelBindingPostProcessor.class); + private final FunctionCatalog functionCatalog; private final FunctionInspector functionInspector; @@ -88,12 +155,14 @@ public class FunctionConfiguration { private GenericApplicationContext context; + FunctionChannelBindingPostProcessor(FunctionCatalog functionCatalog, FunctionInspector functionInspector, StreamFunctionProperties functionProperties, BindableProxyFactory bindableProxyFactory) { this.functionCatalog = functionCatalog; this.functionInspector = functionInspector; this.functionProperties = functionProperties; this.bindableProxyFactory = bindableProxyFactory; + } public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException { @@ -111,14 +180,13 @@ public class FunctionConfiguration { } private void doPostProcess(String channelName, SubscribableChannel messageChannel) { - //TODO there is something about moving channel interceptors in AMCB (not sure if it is still required) if (functionProperties.isComposeTo() && messageChannel instanceof SubscribableChannel && "input".equals(channelName)) { - System.out.println("Composing at the tail"); + throw new UnsupportedOperationException("Composing at tail is not currently supported"); } else if (functionProperties.isComposeFrom() && "output".equals(channelName)) { - Assert.notNull(bindableProxyFactory, "Can not compose function into the existing app since `bindableProxyFactory` is null."); - System.out.println("Composing at the head"); + Assert.notNull(this.bindableProxyFactory, "Can not compose function into the existing app since `bindableProxyFactory` is null."); + logger.info("Composing at the head of 'output' channel"); FunctionInvocationWrapper function = functionCatalog.lookup(functionProperties.getDefinition(), "application/json"); ServiceActivatingHandler handler = new ServiceActivatingHandler(new FunctionWrapper(function)); handler.setBeanFactory(context); @@ -133,20 +201,17 @@ public class FunctionConfiguration { handler.setOutputChannelName("output.extended"); SubscribableChannel subscribeChannel = (SubscribableChannel) messageChannel; subscribeChannel.subscribe(handler); - } else { FunctionInvocationWrapper function = functionCatalog.lookup(functionProperties.getDefinition(), "application/json"); - if (function.getTarget() instanceof Supplier) { - System.out.println("Configuring supplier"); - throw new UnsupportedOperationException("Standalone supplier are not currently supported"); - } - else if (function.getTarget() instanceof Consumer) { - throw new UnsupportedOperationException("Consumers are not currently supported"); - } - else { - if ("input".equals(channelName)) { - this.postProcessForStandAloneFunction(function, messageChannel); + if (!function.isSupplier()) { + if (function.isConsumer()) { + throw new UnsupportedOperationException("Consumers are not currently supported"); + } + else if (function.isFunction()) { + if ("input".equals(channelName)) { + this.postProcessForStandAloneFunction(function, messageChannel); + } } } } @@ -170,12 +235,17 @@ public class FunctionConfiguration { } } + /* + * Enhance publisher to add error handling, retries etc. + */ @SuppressWarnings({ "unchecked", "rawtypes" }) private Publisher enhancePublisher(Publisher publisher) { Flux flux = Flux.from(publisher) .concatMap(message -> { return Flux.just(message) - .doOnError(e -> e.printStackTrace()) + .doOnError(e -> { + e.printStackTrace(); + }) .retryBackoff(3, //this.consumerProperties.getMaxAttempts(), Duration.ofMillis(1000), //this.consumerProperties.getBackOffInitialInterval()), @@ -190,6 +260,7 @@ public class FunctionConfiguration { return flux; } + @SuppressWarnings({ "unchecked", "rawtypes" }) private void subscribeToInput(Function function, Publisher publisher, Consumer> outputProcessor) { diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/IntegrationFlowFunctionSupport.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/IntegrationFlowFunctionSupport.java index 55d22f048..0d54b035c 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/IntegrationFlowFunctionSupport.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/IntegrationFlowFunctionSupport.java @@ -61,7 +61,7 @@ public class IntegrationFlowFunctionSupport { private final StreamFunctionProperties functionProperties; - private final AtomicReference> triggerRef = new AtomicReference<>(); + //private final AtomicReference> triggerRef = new AtomicReference<>(); private final Publisher trigger; @@ -85,6 +85,7 @@ public class IntegrationFlowFunctionSupport { this.functionProperties = functionProperties; this.context = context; this.functionProperties.setBindingServiceProperties(bindingServiceProperties); + AtomicReference> triggerRef = null; trigger = Mono.create(emmiter -> { triggerRef.set(emmiter); }); diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/SourceToFunctionsSupportTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/SourceToFunctionsSupportTests.java index 2b0421007..bb39c06bb 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/SourceToFunctionsSupportTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/SourceToFunctionsSupportTests.java @@ -17,17 +17,13 @@ package org.springframework.cloud.stream.function; import java.nio.charset.StandardCharsets; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Function; import java.util.function.Supplier; -import org.junit.Ignore; import org.junit.Rule; import org.junit.Test; import org.junit.rules.ExpectedException; -import org.reactivestreams.Publisher; import reactor.core.publisher.Flux; import org.springframework.boot.WebApplicationType; @@ -39,16 +35,11 @@ import org.springframework.cloud.stream.binder.test.TestChannelBinderConfigurati 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.channel.QueueChannel; import org.springframework.integration.dsl.IntegrationFlow; 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.messaging.PollableChannel; -import org.springframework.scheduling.annotation.EnableScheduling; -import org.springframework.scheduling.annotation.Scheduled; import org.springframework.util.MimeTypeUtils; import static org.assertj.core.api.Assertions.assertThat; @@ -67,7 +58,7 @@ public class SourceToFunctionsSupportTests { public void testFunctionIsAppliedToExistingMessageSource() { try (ConfigurableApplicationContext context = new SpringApplicationBuilder( TestChannelBinderConfiguration.getCompleteConfiguration( - FunctionsConfiguration.class)).web(WebApplicationType.NONE).run( + FunctionsConfiguration.class, ExistingMessageSourceConfiguration.class)).web(WebApplicationType.NONE).run( "--spring.cloud.stream.function.definition=|toUpperCase", "--spring.jmx.enabled=false")) { @@ -78,165 +69,153 @@ public class SourceToFunctionsSupportTests { } @Test - public void testComposedFunctionIsAppliedToExistingMessageSource() { + public void testFunctionsAreAppliedToExistingMessageSource() { try (ConfigurableApplicationContext context = new SpringApplicationBuilder( TestChannelBinderConfiguration.getCompleteConfiguration( - FunctionsConfiguration.class)).web(WebApplicationType.NONE).run( + FunctionsConfiguration.class, ExistingMessageSourceConfiguration.class)).web(WebApplicationType.NONE).run( "--spring.cloud.stream.function.definition=|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)); + assertThat(target.receive(1000).getPayload()) + .isEqualTo("HELLO FUNCTION:HELLO FUNCTION".getBytes(StandardCharsets.UTF_8)); } } @Test - public void testFailedInputTypeConversion() { + public void testImperativeSupplier() { try (ConfigurableApplicationContext context = new SpringApplicationBuilder( TestChannelBinderConfiguration.getCompleteConfiguration( - FunctionsConfigurationNoConversionPossible.class)) - .web(WebApplicationType.NONE) - .run("--spring.cloud.stream.function.definition=|toUpperCase|concatWithSelf", - "--spring.jmx.enabled=false")) { - PollableChannel errorChannel = context.getBean("errorChannel", - PollableChannel.class); - OutputDestination target = context.getBean(OutputDestination.class); - assertThat(target.receive(1000)).isNull(); - assertThat(errorChannel.receive(10000)).isNotNull(); - } - } - - @Test - public void testComposedFunctionIsAppliedToExistingMessageSourceFailedTypeConversion() { - try (ConfigurableApplicationContext context = new SpringApplicationBuilder( - TestChannelBinderConfiguration.getCompleteConfiguration( - FunctionsConfigurationNoConversionPossible.class)) - .web(WebApplicationType.NONE) - .run("--spring.cloud.stream.function.definition=|toUpperCase|concatWithSelf", - "--spring.jmx.enabled=false")) { - PollableChannel errorChannel = context.getBean("errorChannel", - PollableChannel.class); - OutputDestination target = context.getBean(OutputDestination.class); - assertThat(target.receive(1000)).isNull(); - assertThat(errorChannel.receive(10000)).isNotNull(); - } - } - - @Test - @Ignore - public void testMessageSourceIsCreatedFromProvidedSupplier() { - try (ConfigurableApplicationContext context = new SpringApplicationBuilder( - TestChannelBinderConfiguration - .getCompleteConfiguration(SupplierConfiguration.class)) - .web(WebApplicationType.NONE) - .run("--spring.cloud.stream.function.definition=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 - @Ignore - public void testMessageSourceIsCreatedFromProvidedSupplierComposedWithSingleFunction() { - try (ConfigurableApplicationContext context = new SpringApplicationBuilder( - TestChannelBinderConfiguration.getCompleteConfiguration( - SupplierConfiguration.class)).web(WebApplicationType.NONE).run( - "--spring.cloud.stream.function.definition=number|concatWithSelf", + FunctionsConfiguration.class, SupplierConfiguration.class)).web(WebApplicationType.NONE).run( + "--spring.cloud.stream.function.definition=number", "--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 + String result = new String(target.receive(1000).getPayload(), StandardCharsets.UTF_8); + assertThat(result).isEqualTo("1"); } } @Test - @Ignore - public void testMessageSourceIsCreatedFromProvidedSupplierComposedWithMultipleFunctions() { + public void testImperativeSupplierComposedWithFunctions() { try (ConfigurableApplicationContext context = new SpringApplicationBuilder( TestChannelBinderConfiguration.getCompleteConfiguration( - SupplierConfiguration.class)).web(WebApplicationType.NONE).run( - "--spring.cloud.stream.function.definition=number|concatWithSelf|multiplyByTwo", + FunctionsConfiguration.class, SupplierConfiguration.class)).web(WebApplicationType.NONE).run( + "--spring.cloud.stream.function.definition=number|toUpperCase|concatWithSelf", "--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 + String result = new String(target.receive(1000).getPayload(), StandardCharsets.UTF_8); + assertThat(result).isEqualTo("1:1"); } } -// @Test -// public void testMessageSourceIsCreatedFromProvidedStreamSupplier() { -// try (ConfigurableApplicationContext context = new SpringApplicationBuilder( -// TestChannelBinderConfiguration.getCompleteConfiguration( -// StreamSupplierConfiguration.class)).web(WebApplicationType.NONE).run( -// "--spring.cloud.stream.function.definition=stream", -// "--spring.jmx.enabled=false")) { -// -// OutputDestination target = context.getBean(OutputDestination.class); -// assertThat(target.receive(1000).getPayload()) -// .isEqualTo("0".getBytes(StandardCharsets.UTF_8)); -// assertThat(target.receive(1000).getPayload()) -// .isEqualTo("1".getBytes(StandardCharsets.UTF_8)); -// assertThat(target.receive(1000).getPayload()) -// .isEqualTo("2".getBytes(StandardCharsets.UTF_8)); -// assertThat(target.receive(1000).getPayload()) -// .isEqualTo("3".getBytes(StandardCharsets.UTF_8)); -// -// // etc -// } -// } + @Test + public void testImperativeSupplierComposedWithMixedFunctions() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration( + FunctionsConfiguration.class, SupplierConfiguration.class)).web(WebApplicationType.NONE).run( + "--spring.cloud.stream.function.definition=number|toUpperCaseReactive|concatWithSelf", + "--spring.jmx.enabled=false")) { + + OutputDestination target = context.getBean(OutputDestination.class); + String result = new String(target.receive(1000).getPayload(), StandardCharsets.UTF_8); + assertThat(result).isEqualTo("1:1"); + } + } + + @Test + public void testReactiveSupplier() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration(SupplierConfiguration.class)).web(WebApplicationType.NONE).run( + "--spring.cloud.stream.function.definition=numberReactive", + "--spring.jmx.enabled=false")) { + + OutputDestination target = context.getBean(OutputDestination.class); + String result = new String(target.receive(1000).getPayload(), StandardCharsets.UTF_8); + assertThat(result).isEqualTo("0"); + result = new String(target.receive(1000).getPayload(), StandardCharsets.UTF_8); + assertThat(result).isEqualTo("1"); + result = new String(target.receive(1000).getPayload(), StandardCharsets.UTF_8); + assertThat(result).isEqualTo("2"); + } + } + + @Test + public void testReactiveSupplierComposedWithImperativeFunctions() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration(FunctionsConfiguration.class, + SupplierConfiguration.class)).web(WebApplicationType.NONE).run( + "--spring.cloud.stream.function.definition=numberReactive|toUpperCase|concatWithSelf", + "--spring.jmx.enabled=false")) { + + OutputDestination target = context.getBean(OutputDestination.class); + String result = new String(target.receive(1000).getPayload(), StandardCharsets.UTF_8); + assertThat(result).isEqualTo("0:0"); + result = new String(target.receive(1000).getPayload(), StandardCharsets.UTF_8); + assertThat(result).isEqualTo("1:1"); + result = new String(target.receive(1000).getPayload(), StandardCharsets.UTF_8); + assertThat(result).isEqualTo("2:2"); + } + } + + @Test + public void testReactiveSupplierComposedWithMixedFunctions() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration(FunctionsConfiguration.class, + SupplierConfiguration.class)).web(WebApplicationType.NONE).run( + "--spring.cloud.stream.function.definition=numberReactive|concatWithSelf|toUpperCaseReactive", + "--spring.jmx.enabled=false")) { + + OutputDestination target = context.getBean(OutputDestination.class); + String result = new String(target.receive(1000).getPayload(), StandardCharsets.UTF_8); + assertThat(result).isEqualTo("0:0"); + result = new String(target.receive(1000).getPayload(), StandardCharsets.UTF_8); + assertThat(result).isEqualTo("1:1"); + result = new String(target.receive(1000).getPayload(), StandardCharsets.UTF_8); + assertThat(result).isEqualTo("2:2"); + } + } + + @Test + public void testReactiveSupplierComposedWithMixedFunctions2() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration(FunctionsConfiguration.class, + SupplierConfiguration.class)).web(WebApplicationType.NONE).run( + "--spring.cloud.stream.function.definition=numberReactive|toUpperCaseReactive|concatWithSelf", + "--spring.jmx.enabled=false")) { + + OutputDestination target = context.getBean(OutputDestination.class); + String result = new String(target.receive(1000).getPayload(), StandardCharsets.UTF_8); + assertThat(result).isEqualTo("0:0"); + result = new String(target.receive(1000).getPayload(), StandardCharsets.UTF_8); + assertThat(result).isEqualTo("1:1"); + result = new String(target.receive(1000).getPayload(), StandardCharsets.UTF_8); + assertThat(result).isEqualTo("2:2"); + } + } -// @Test -// public void testFunctionDoesNotExist() { -// -// this.expectedException.expect(BeanCreationException.class); -// -// new SpringApplicationBuilder(TestChannelBinderConfiguration -// .getCompleteConfiguration(SupplierConfiguration.class)) -// .web(WebApplicationType.NONE) -// .run("--spring.cloud.stream.function.definition=doesNotExist", -// "--spring.jmx.enabled=false"); -// } @EnableAutoConfiguration - //@Import(ProvidedMessageSourceConfiguration.class) - @EnableScheduling public static class SupplierConfiguration { AtomicInteger counter = new AtomicInteger(); @Bean - @Scheduled(fixedRate = 5000) public Supplier number() { - return () -> { - return String.valueOf(this.counter.incrementAndGet()); - }; + return () -> String.valueOf(this.counter.incrementAndGet()); } @Bean - public Function concatWithSelf() { - return x -> x + x; + public Supplier> numberReactive() { + return () -> Flux.create(emitter -> { + for (int i = 0; i < 3; i++) { + emitter.next(String.valueOf(i)); + } + }); } + @Bean public Function, Flux> multiplyByTwo() { return x -> x.map(i -> String.valueOf(Integer.valueOf(i) * 2)); @@ -245,7 +224,6 @@ public class SourceToFunctionsSupportTests { } @EnableAutoConfiguration - @Import(ExistingMessageSourceConfiguration.class) public static class FunctionsConfiguration { @Bean @@ -253,6 +231,11 @@ public class SourceToFunctionsSupportTests { return String::toUpperCase; } + @Bean + public Function, Flux> toUpperCaseReactive() { + return flux -> flux.map(String::toUpperCase); + } + @Bean public Function concatWithSelf() { return x -> x + ":" + x; @@ -260,44 +243,6 @@ public class SourceToFunctionsSupportTests { } - @EnableAutoConfiguration - public static class StreamSupplierConfiguration { - - @Bean - public Supplier> stream() { - ExecutorService executor = Executors.newFixedThreadPool(1); - return () -> Flux.create(emitter -> { - executor.execute(() -> { - for (int i = 0; i < 10; i++) { - emitter.next(MessageBuilder.withPayload(String.valueOf(i)).build()); - } - }); - }); - } - - } - - @EnableAutoConfiguration - @Import(ExistingMessageSourceConfigurationNoContentTypeSet.class) - public static class FunctionsConfigurationNoConversionPossible { - - @Bean - public PollableChannel errorChannel() { - return new QueueChannel(10); - } - - @Bean - public Function toUpperCase() { - return x -> true; - } - - @Bean - public Function concatWithSelf() { - return x -> 1; - } - - } - /** * This configuration essentially emulates our existing app-starters for Sources and * essentially demonstrates how a function(s) could be applied to an existing source @@ -306,18 +251,12 @@ public class SourceToFunctionsSupportTests { @EnableBinding(Source.class) public static class ExistingMessageSourceConfiguration { -// @Autowired -// private Source source; - @Bean public IntegrationFlow messageSourceFlow() { Supplier> messageSource = () -> MessageBuilder .withPayload("hello function") .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN) .build(); - -// return functionSupport.integrationFlowFromProvidedSupplier(messageSource) -// .channel(this.source.output()).get(); return IntegrationFlows.from(messageSource).channel("output").get(); } @@ -332,33 +271,9 @@ public class SourceToFunctionsSupportTests { .withPayload("hello function") .setHeader(MessageHeaders.CONTENT_TYPE, "application/octet-stream") .build(); - -// return functionSupport.integrationFlowFromProvidedSupplier(messageSource) -// .channel(this.source.output()).get(); return IntegrationFlows.from(messageSource).channel("output").get(); } } -// @EnableBinding(Source.class) -// public static class ProvidedMessageSourceConfiguration { -// -// @Autowired -// private Source source; -// -// @Autowired -// private StreamFunctionProperties functionProperties; -// -// @Bean -// public IntegrationFlow messageSourceFlow( -// IntegrationFlowFunctionSupport functionSupport) { -// Assert.hasText(this.functionProperties.getDefinition(), -// "Supplier name must be provided"); -// -// return functionSupport.integrationFlowFromNamedSupplier() -// .channel(this.source.output()).get(); -// } -// -// } - }