diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java index 913e0e175..643d3763a 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java @@ -44,6 +44,8 @@ import org.springframework.cloud.stream.binding.InputBindingLifecycle; import org.springframework.cloud.stream.binding.MessageChannelStreamListenerResultAdapter; import org.springframework.cloud.stream.binding.OutputBindingLifecycle; import org.springframework.cloud.stream.binding.StreamListenerAnnotationBeanPostProcessor; +import org.springframework.cloud.stream.function.FunctionConfiguration; +import org.springframework.cloud.stream.function.FunctionProperties; import org.springframework.cloud.stream.micrometer.DestinationPublishingMetricsAutoConfiguration; import org.springframework.context.ApplicationListener; import org.springframework.context.annotation.Bean; @@ -76,8 +78,8 @@ import org.springframework.util.Assert; * @author Soby Chacko */ @Configuration -@EnableConfigurationProperties({ BindingServiceProperties.class, SpringIntegrationProperties.class }) -@Import({DestinationPublishingMetricsAutoConfiguration.class, SpelExpressionConverterConfiguration.class}) +@EnableConfigurationProperties({ BindingServiceProperties.class, SpringIntegrationProperties.class, FunctionProperties.class }) +@Import({ DestinationPublishingMetricsAutoConfiguration.class, SpelExpressionConverterConfiguration.class, FunctionConfiguration.class }) @Role(BeanDefinition.ROLE_INFRASTRUCTURE) @ConditionalOnBean(value = BinderTypeRegistry.class, search = SearchStrategy.CURRENT) public class BindingServiceConfiguration { @@ -91,7 +93,8 @@ public class BindingServiceConfiguration { @Bean @ConditionalOnMissingBean(BinderFactory.class) public BinderFactory binderFactory(BinderTypeRegistry binderTypeRegistry, - BindingServiceProperties bindingServiceProperties) { + BindingServiceProperties bindingServiceProperties) { + DefaultBinderFactory binderFactory = new DefaultBinderFactory( getBinderConfigurations(binderTypeRegistry, bindingServiceProperties), binderTypeRegistry); binderFactory.setDefaultBinder(bindingServiceProperties.getDefaultBinder()); @@ -100,7 +103,8 @@ public class BindingServiceConfiguration { } private static Map getBinderConfigurations(BinderTypeRegistry binderTypeRegistry, - BindingServiceProperties bindingServiceProperties) { + BindingServiceProperties bindingServiceProperties) { + Map binderConfigurations = new HashMap<>(); Map declaredBinders = bindingServiceProperties.getBinders(); boolean defaultCandidatesExist = false; @@ -161,12 +165,15 @@ public class BindingServiceConfiguration { @ConditionalOnMissingBean(search = SearchStrategy.CURRENT) public BindingService bindingService(BindingServiceProperties bindingServiceProperties, BinderFactory binderFactory, TaskScheduler taskScheduler) { + return new BindingService(bindingServiceProperties, binderFactory, taskScheduler); } @Bean @DependsOn("bindingService") - public OutputBindingLifecycle outputBindingLifecycle(BindingService bindingService, Map bindables) { + public OutputBindingLifecycle outputBindingLifecycle(BindingService bindingService, + Map bindables) { + return new OutputBindingLifecycle(bindingService, bindables); } @@ -189,6 +196,7 @@ public class BindingServiceConfiguration { DynamicDestinationsBindable dynamicDestinationsBindable, @Nullable BinderAwareChannelResolver.NewDestinationBindingCallback callback, @Nullable GlobalChannelInterceptorProcessor globalChannelInterceptorProcessor) { + return new BinderAwareChannelResolver(bindingService, bindingTargetFactory, dynamicDestinationsBindable, callback, globalChannelInterceptorProcessor); } @@ -202,8 +210,8 @@ public class BindingServiceConfiguration { @Bean @ConditionalOnMissingBean public org.springframework.cloud.stream.binding.BinderAwareRouterBeanPostProcessor binderAwareRouterBeanPostProcessor( - @Autowired(required=false) AbstractMappingMessageRouter[] routers, - @Autowired(required=false)DestinationResolver channelResolver) { + @Autowired(required = false) AbstractMappingMessageRouter[] routers, + @Autowired(required = false) DestinationResolver channelResolver) { return new org.springframework.cloud.stream.binding.BinderAwareRouterBeanPostProcessor(routers, channelResolver); } @@ -211,11 +219,15 @@ public class BindingServiceConfiguration { @Bean public ApplicationListener appListener(SpringIntegrationProperties springIntegrationProperties) { return new ApplicationListener() { + @Override public void onApplicationEvent(ContextRefreshedEvent event) { event.getApplicationContext().getBeansOfType(AbstractReplyProducingMessageHandler.class).values() - .forEach(mh -> mh.addNotPropagatedHeaders(springIntegrationProperties.getMessageHandlerNotPropagatedHeaders())); + .forEach(mh -> mh.addNotPropagatedHeaders(springIntegrationProperties + .getMessageHandlerNotPropagatedHeaders())); } + }; } + } 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 new file mode 100644 index 000000000..797aaa9aa --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java @@ -0,0 +1,43 @@ +/* + * 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 org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +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.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +/** + * + * @author Oleg Zhurakousky + * + * @since 2.1 + */ +@Configuration +@ConditionalOnProperty("spring.cloud.stream.function.name") +public class FunctionConfiguration { + + @Bean + public FunctionSupport functionSupport(FunctionCatalog functionCatalog, FunctionInspector functionInspector, + CompositeMessageConverterFactory messageConverterFactory) { + + return new FunctionSupport(functionCatalog, functionInspector, messageConverterFactory); + } + +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionInvoker.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionInvoker.java new file mode 100644 index 000000000..266bb5abe --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionInvoker.java @@ -0,0 +1,101 @@ +/* + * 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.concurrent.atomic.AtomicReference; +import java.util.function.Function; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import reactor.core.publisher.Flux; + +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.stream.converter.CompositeMessageConverterFactory; +import org.springframework.messaging.Message; +import org.springframework.messaging.converter.CompositeMessageConverter; +import org.springframework.util.Assert; + +/** + * + * @author Oleg Zhurakousky + * + * @param the payload type of the input Message + * @param the payload type of the output Message + * + * @since 2.1 + */ +class FunctionInvoker implements Function>, Flux>> { + + private static final Log logger = LogFactory.getLog(FunctionInvoker.class); + + private final Class inputClass; + + private final Function, Flux> userFunction; + + private final CompositeMessageConverter messageConverter; + + FunctionInvoker(String functionName, FunctionCatalog functionCatalog, FunctionInspector functionInspector, + CompositeMessageConverterFactory compositeMessageConverterFactory) { + + this.userFunction = functionCatalog.lookup(Function.class, functionName); + Assert.notNull(this.userFunction, "userFunction: " + functionName + " can not be located."); + this.messageConverter = compositeMessageConverterFactory.getMessageConverterForAllRegistered(); + FunctionType functionType = functionInspector.getRegistration(this.userFunction).getType(); + this.inputClass = functionType.getInputType(); + } + + @Override + public Flux> apply(Flux> input) { + AtomicReference> originalMessage = new AtomicReference<>(); + return input + .doOnNext(originalMessage::set) // to preserve the original message + .map(this::resolveArgument) // resolves argument type before invocation of user function + .transform(this.userFunction::apply) // invoke user function + .map(resultMessage -> toMessage(resultMessage, originalMessage.get())); // create output message + } + + @SuppressWarnings("unchecked") + private Message toMessage(T value, Message originalMessage) { + if (logger.isDebugEnabled()) { + logger.debug("Converting result back to message using the original message: " + originalMessage); + } + return (Message) + (value instanceof Message + ? value + : this.messageConverter.toMessage(value, originalMessage.getHeaders())); + } + + @SuppressWarnings("unchecked") + private T resolveArgument(Message message) { + if (logger.isDebugEnabled()) { + logger.debug("Resolving input argument from message: " + message); + } + + return (T) (shouldConvertFromMessage(message) + ? this.messageConverter.fromMessage(message, this.inputClass) + : message); + } + + private boolean shouldConvertFromMessage(Message message) { + return !this.inputClass.isAssignableFrom(byte[].class) && + !this.inputClass.isAssignableFrom(Object.class); + } + +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionProperties.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionProperties.java new file mode 100644 index 000000000..39488dae1 --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionProperties.java @@ -0,0 +1,44 @@ +/* + * 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 org.springframework.boot.context.properties.ConfigurationProperties; + +/** + * + * @author Oleg Zhurakousky + * + * @since 2.1 + */ +@ConfigurationProperties("spring.cloud.stream.function") +public class FunctionProperties { + + /** + * Name of functions to bind. If several functions need to be composed into one, use pipes (e.g., 'fooFunc|barFunc') + */ + private String name; + + + public String getName() { + return this.name; + } + + public void setName(String name) { + this.name = name; + } + +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionSupport.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionSupport.java new file mode 100644 index 000000000..e5d261322 --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionSupport.java @@ -0,0 +1,86 @@ +/* + * 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 void applyFunctionToIntegrationFlow(String functionName, IntegrationFlowBuilder flowBuilder, + Consumer> outputProcessor) { + + FunctionInvoker functionInvoker = + new FunctionInvoker<>(functionName, this.functionCatalog, this.functionInspector, + this.messageConverterFactory); + + subscribeToInput(functionInvoker, flowBuilder.toReactivePublisher(), outputProcessor); + } + + private Mono subscribeToOutput(Consumer> outputProcessor, + Publisher> outputPublisher) { + + Flux> output = outputProcessor == null + ? Flux.from(outputPublisher) + : Flux.from(outputPublisher).doOnNext(outputProcessor); + return output.then(); + } + + private void subscribeToInput(FunctionInvoker functionInvoker, Publisher> publisher, + Consumer> outputProcessor) { + + Flux> inputPublisher = Flux.from(publisher); + subscribeToOutput(outputProcessor, functionInvoker.apply(inputPublisher)).subscribe(); + } + +} diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/FunctionSupportTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/FunctionSupportTests.java new file mode 100644 index 000000000..4564fa1b1 --- /dev/null +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/FunctionSupportTests.java @@ -0,0 +1,137 @@ +/* + * 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.TimeUnit; +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.scheduling.PollerMetadata; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHeaders; +import org.springframework.scheduling.support.PeriodicTrigger; +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 toUpperCase() { + return String::toUpperCase; + } + + @Bean + public Function 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> 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(); + } + + } + +}