diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractBinder.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractBinder.java index e4fd06db8..22a09e531 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractBinder.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractBinder.java @@ -49,7 +49,7 @@ import org.springframework.util.StringUtils; * @author Oleg Zhurakousky */ public abstract class AbstractBinder - implements ApplicationContextAware, InitializingBean, Binder { + implements ApplicationContextAware, InitializingBean, Binder { /** * The delimiter between a group and index when constructing a binder @@ -61,10 +61,9 @@ public abstract class AbstractBinder binding = new DefaultBinding(destination, outputChannel, producerMessageHandler instanceof Lifecycle ? (Lifecycle) producerMessageHandler : null) { @@ -207,6 +217,8 @@ public abstract class AbstractMessageChannelBinder doGetExtendedInfo(Object destination, Object properties) { Map extendedInfo = new LinkedHashMap<>(); extendedInfo.put("bindingDestination", destination.toString()); diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/BinderFunctionSupport.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/BinderFunctionSupport.java new file mode 100644 index 000000000..fae10d95b --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/BinderFunctionSupport.java @@ -0,0 +1,42 @@ +/* + * 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.Function; + +import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.channel.MessageChannelReactiveUtils; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.SubscribableChannel; + +/** + * @author David Turanski + * @since 2.1 + **/ +public abstract class BinderFunctionSupport { + + public static SubscribableChannel andThenFunctionDefinition(IntegrationFlowFunctionSupport + integrationFlowFunctionSupport, MessageChannel outputChannel) { + if (integrationFlowFunctionSupport != null && integrationFlowFunctionSupport.containsFunction(Function.class)) { + DirectChannel actualOutputChannel = new DirectChannel(); + integrationFlowFunctionSupport.andThenFunction(MessageChannelReactiveUtils.toPublisher(outputChannel), + actualOutputChannel); + return actualOutputChannel; + } + return (SubscribableChannel) outputChannel; + } +} 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 8572b92ef..f5973df9c 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 @@ -32,30 +32,29 @@ import org.springframework.context.annotation.Configuration; import org.springframework.integration.dsl.IntegrationFlow; /** - * * @author Oleg Zhurakousky - * + * @author David Turanski * @since 2.1 */ @Configuration @ConditionalOnProperty("spring.cloud.stream.function.definition") public class FunctionConfiguration { - @Autowired(required=false) + @Autowired(required = false) private Source source; - @Autowired(required=false) + @Autowired(required = false) private Processor processor; - @Autowired(required=false) + @Autowired(required = false) private Sink sink; @Bean public IntegrationFlowFunctionSupport functionSupport(FunctionCatalogWrapper functionCatalog, - FunctionInspector functionInspector, CompositeMessageConverterFactory messageConverterFactory, - StreamFunctionProperties functionProperties) { + FunctionInspector functionInspector, CompositeMessageConverterFactory messageConverterFactory, + StreamFunctionProperties functionProperties) { return new IntegrationFlowFunctionSupport(functionCatalog, functionInspector, messageConverterFactory, - functionProperties); + functionProperties); } @Bean @@ -67,7 +66,8 @@ public class FunctionConfiguration { * This configuration creates an instance of {@link IntegrationFlow} appropriate for binding declared using EnableBinding. * At the moment only Source, Processor and Sink are supported. */ - @ConditionalOnMissingBean // starter apps typically already provide and instance of IntegrationFlow, so we don't need this one. + @ConditionalOnMissingBean + // starter apps typically already provide and instance of IntegrationFlow, so we don't need this one. @Bean public IntegrationFlow integrationFlowCreator(IntegrationFlowFunctionSupport functionSupport) { if (processor != null) { @@ -77,10 +77,11 @@ public class FunctionConfiguration { return functionSupport.integrationFlowForFunction(sink.input(), null).get(); } else if (source != null) { - return functionSupport.containsFunction(Supplier.class) - ? functionSupport.integrationFlowFromNamedSupplier().channel(this.source.output()).get() - : null; + return functionSupport.containsFunction(Supplier.class) ? + functionSupport.integrationFlowFromNamedSupplier().channel(this.source.output()).get() : + null; } - throw new UnsupportedOperationException("Bindings other then Source, Processor and Sink are not currently supported"); + throw new UnsupportedOperationException( + "Bindings other then Source, Processor and Sink are not currently supported"); } } 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 e51e96e44..70ddee1a4 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 @@ -21,17 +21,19 @@ 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.beans.factory.annotation.Autowired; +import org.springframework.beans.BeansException; +import org.springframework.beans.factory.InitializingBean; import org.springframework.cloud.function.context.FunctionCatalog; import org.springframework.cloud.function.context.FunctionType; 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.channel.FluxMessageChannel; +import org.springframework.context.ApplicationContext; +import org.springframework.context.ApplicationContextAware; +import org.springframework.integration.context.IntegrationContextUtils; import org.springframework.integration.dsl.IntegrationFlowBuilder; import org.springframework.integration.dsl.IntegrationFlows; import org.springframework.messaging.Message; @@ -44,10 +46,11 @@ import org.springframework.util.StringUtils; * * @author Oleg Zhurakousky * @author David Turanski + * @author Ilayaperumal Gopinathan * * @since 2.1 */ -public class IntegrationFlowFunctionSupport { +public class IntegrationFlowFunctionSupport implements ApplicationContextAware, InitializingBean { private final FunctionCatalogWrapper functionCatalog; @@ -57,8 +60,8 @@ public class IntegrationFlowFunctionSupport { private final StreamFunctionProperties functionProperties; - @Autowired private MessageChannel errorChannel; + private ApplicationContext applicationContext; /** * @param functionCatalog @@ -92,7 +95,8 @@ public class IntegrationFlowFunctionSupport { } public FunctionType getCurrentFunctionType() { - FunctionType functionType = functionInspector.getRegistration(functionCatalog.lookup(this.functionProperties.getDefinition())).getType(); + FunctionType functionType = functionInspector.getRegistration( + functionCatalog.lookup(this.functionProperties.getDefinition())).getType(); return functionType; } @@ -161,33 +165,20 @@ public class IntegrationFlowFunctionSupport { * @return true if {@link Function} was located and added and false if it wasn't. */ public boolean andThenFunction(IntegrationFlowBuilder flowBuilder, MessageChannel outputChannel) { - if (StringUtils.hasText(this.functionProperties.getDefinition())) { - FunctionInvoker functionInvoker = - new FunctionInvoker<>(this.functionProperties.getDefinition(), this.functionCatalog, - this.functionInspector, this.messageConverterFactory, this.errorChannel); - - if (outputChannel != null) { - subscribeToInput(functionInvoker, flowBuilder.toReactivePublisher(), outputChannel::send); - } - else { - subscribeToInput(functionInvoker, flowBuilder.toReactivePublisher(), null); - } - return true; - } - return false; + return andThenFunction(flowBuilder.toReactivePublisher(), outputChannel); } - public boolean andThenFunction(FluxMessageChannel fluxChannel, MessageChannel outputChannel) { + public boolean andThenFunction(Publisher publisher, MessageChannel outputChannel) { if (StringUtils.hasText(this.functionProperties.getDefinition())) { - FunctionInvoker functionInvoker = + FunctionInvoker functionInvoker = new FunctionInvoker<>(this.functionProperties.getDefinition(), this.functionCatalog, this.functionInspector, this.messageConverterFactory, this.errorChannel); if (outputChannel != null) { - subscribeToInput(functionInvoker, fluxChannel, outputChannel::send); + subscribeToInput(functionInvoker, publisher, outputChannel::send); } else { - subscribeToInput(functionInvoker, fluxChannel, null); + subscribeToInput(functionInvoker, publisher, null); } return true; } @@ -211,4 +202,13 @@ public class IntegrationFlowFunctionSupport { subscribeToOutput(outputProcessor, functionInvoker.apply((Flux>) inputPublisher)).subscribe(); } + @Override + public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { + this.applicationContext = applicationContext; + } + + @Override + public void afterPropertiesSet() throws Exception { + this.errorChannel = (MessageChannel) applicationContext.getBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME); + } } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ProcessorToFunctionsSupportTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ProcessorToFunctionsSupportTests.java index d74602d66..932d4a93a 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ProcessorToFunctionsSupportTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ProcessorToFunctionsSupportTests.java @@ -16,12 +16,11 @@ package org.springframework.cloud.stream.function; - - import java.nio.charset.StandardCharsets; import java.util.function.Consumer; import java.util.function.Function; +import org.junit.After; import org.junit.Test; import org.springframework.beans.DirectFieldAccessor; @@ -34,34 +33,35 @@ import org.springframework.cloud.stream.binder.test.InputDestination; import org.springframework.cloud.stream.binder.test.OutputDestination; import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration; import org.springframework.cloud.stream.messaging.Processor; -import org.springframework.context.ApplicationContext; +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.dsl.IntegrationFlows; -import org.springframework.lang.Nullable; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.support.GenericMessage; import static org.assertj.core.api.Assertions.assertThat; /** - * * @author Oleg Zhurakousky - * + * @author David Turanski * @since 2.1 - * */ public class ProcessorToFunctionsSupportTests { + private ConfigurableApplicationContext context; + + @After + public void cleanUp() { + context.close(); + } + @Test public void testPathThrough() { - ApplicationContext context = - new SpringApplicationBuilder( - TestChannelBinderConfiguration.getCompleteConfiguration(FunctionsConfiguration.class)) - .web(WebApplicationType.NONE) - .run("--spring.jmx.enabled=false"); + context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration(FunctionsConfiguration.class)).web( + WebApplicationType.NONE).run("--spring.jmx.enabled=false"); InputDestination source = context.getBean(InputDestination.class); OutputDestination target = context.getBean(OutputDestination.class); source.send(new GenericMessage("hello".getBytes(StandardCharsets.UTF_8))); @@ -70,11 +70,10 @@ public class ProcessorToFunctionsSupportTests { @Test public void testSingleFunction() { - ApplicationContext context = - new SpringApplicationBuilder( - TestChannelBinderConfiguration.getCompleteConfiguration(FunctionsConfiguration.class)) - .web(WebApplicationType.NONE) - .run("--spring.cloud.stream.function.definition=toUpperCase", "--spring.jmx.enabled=false"); + context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration(FunctionsConfiguration.class)).web( + WebApplicationType.NONE) + .run("--spring.cloud.stream.function.definition=toUpperCase", "--spring.jmx.enabled=false"); InputDestination source = context.getBean(InputDestination.class); OutputDestination target = context.getBean(OutputDestination.class); @@ -84,11 +83,11 @@ public class ProcessorToFunctionsSupportTests { @Test public void testComposedFunction() { - ApplicationContext context = - new SpringApplicationBuilder( - TestChannelBinderConfiguration.getCompleteConfiguration(FunctionsConfiguration.class)) - .web(WebApplicationType.NONE) - .run("--spring.cloud.stream.function.definition=toUpperCase|concatWithSelf", "--spring.jmx.enabled=false"); + context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration(FunctionsConfiguration.class)).web( + WebApplicationType.NONE) + .run("--spring.cloud.stream.function.definition=toUpperCase|concatWithSelf", + "--spring.jmx" + ".enabled=false", "--logging.level.org.springframework.integration=DEBUG"); InputDestination source = context.getBean(InputDestination.class); OutputDestination target = context.getBean(OutputDestination.class); @@ -98,11 +97,9 @@ public class ProcessorToFunctionsSupportTests { @Test public void testConsumer() { - ApplicationContext context = - new SpringApplicationBuilder( - TestChannelBinderConfiguration.getCompleteConfiguration(ConsumerConfiguration.class)) - .web(WebApplicationType.NONE) - .run("--spring.cloud.stream.function.definition=log", "--spring.jmx.enabled=false"); + context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration(ConsumerConfiguration.class)).web( + WebApplicationType.NONE).run("--spring.cloud.stream.function.definition=log", "--spring.jmx.enabled=false"); InputDestination source = context.getBean(InputDestination.class); OutputDestination target = context.getBean(OutputDestination.class); @@ -158,20 +155,10 @@ public class ProcessorToFunctionsSupportTests { private Processor processor; @Bean - public IntegrationFlow fromChannel(@Nullable IntegrationFlowFunctionSupport functionSupport) { + public IntegrationFlow fromChannel() { - IntegrationFlowBuilder flowBuilder = null; - if (functionSupport == null) { - flowBuilder = IntegrationFlows.from(processor.input()).bridge().channel(processor.output()); - } - else { - flowBuilder = functionSupport.integrationFlowFromChannel(processor.input()); - if (!functionSupport.andThenFunction(flowBuilder, processor.output())) { - flowBuilder = flowBuilder.channel(processor.output()); - } - } - - return flowBuilder.get(); + return IntegrationFlows.from(processor.input()) + .channel(processor.output()).get(); } } 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 ae4a70c4b..772a5a93c 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 @@ -41,7 +41,6 @@ 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.IntegrationFlowBuilder; import org.springframework.integration.support.MessageBuilder; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHeaders; @@ -84,7 +83,8 @@ public class SourceToFunctionsSupportTests { try (ConfigurableApplicationContext context = new SpringApplicationBuilder( TestChannelBinderConfiguration.getCompleteConfiguration(FunctionsConfiguration.class)).web( WebApplicationType.NONE) - .run("--spring.cloud.stream.function.definition=toUpperCase|concatWithSelf", "--spring.jmx.enabled=false")) { + .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)); @@ -93,12 +93,11 @@ public class SourceToFunctionsSupportTests { @Test public void testFailedInputTypeConversion() { - try (ConfigurableApplicationContext context = - new SpringApplicationBuilder( - TestChannelBinderConfiguration.getCompleteConfiguration(FunctionsConfigurationNoConversionPossible.class)) - .web(WebApplicationType.NONE) - .run("--spring.cloud.stream.function.definition=toUpperCase|concatWithSelf", - "--spring.jmx.enabled=false")) { + 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); assertNull(target.receive(1000)); @@ -108,12 +107,11 @@ public class SourceToFunctionsSupportTests { @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")) { + 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); assertNull(target.receive(1000)); @@ -125,7 +123,8 @@ public class SourceToFunctionsSupportTests { 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")) { + 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)); @@ -257,13 +256,9 @@ public class SourceToFunctionsSupportTests { .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(); + return functionSupport.integrationFlowFromProvidedSupplier(messageSource) + .channel(this.source.output()) + .get(); } } @@ -277,15 +272,12 @@ public class SourceToFunctionsSupportTests { @Bean public IntegrationFlow messageSourceFlow(IntegrationFlowFunctionSupport functionSupport) { Supplier> messageSource = () -> MessageBuilder.withPayload("hello function") - .setHeader(MessageHeaders.CONTENT_TYPE, "application/octet-stream").build(); + .setHeader(MessageHeaders.CONTENT_TYPE, "application/octet-stream") + .build(); - IntegrationFlowBuilder flowBuilder = functionSupport.integrationFlowFromProvidedSupplier(messageSource); - - if (!functionSupport.andThenFunction(flowBuilder, this.source.output())) { - flowBuilder = flowBuilder.channel(this.source.output()); - } - - return flowBuilder.get(); + return functionSupport.integrationFlowFromProvidedSupplier(messageSource) + .channel(this.source.output()) + .get(); } }