diff --git a/docs/src/main/asciidoc/spring-cloud-stream.adoc b/docs/src/main/asciidoc/spring-cloud-stream.adoc index 4386a7da5..e9a46d181 100644 --- a/docs/src/main/asciidoc/spring-cloud-stream.adoc +++ b/docs/src/main/asciidoc/spring-cloud-stream.adoc @@ -542,8 +542,8 @@ So, this is where understanding of the naming convention for binding destinatio *Binding naming convention:* -* input - ` + _in_ + ` -* output - ` + _out_ + ` +* input - ` + .in. + ` +* output - ` + .out. + ` Let's look at the few samples: @@ -566,15 +566,15 @@ public class SampleApplication { The above example demonstrates function which takes two inputs (first of type `String` and second of type `Integer`) and produces a single output of type `String`. -So, for the above example the two input bindings will be `gather_in_0` and `gather_in_1` and for consistency the -output binding also follows the same convention and is named `gather_out_0`. +So, for the above example the two input bindings will be `gather.in.0` and `gather.in.1` and for consistency the +output binding also follows the same convention and is named `gather.out.0`. Knowing that will allow you to set binding specific properties the same way you did with `@StreamListener`. -For example, the following will override content-type for `gather_in_0` binding: +For example, the following will override content-type for `gather.in.0` binding: ---- ---spring.cloud.stream.bindings.gather_in_0.content-type=text/plain +--spring.cloud.stream.bindings.gather.in.0.content-type=text/plain ---- @@ -601,8 +601,8 @@ public class SampleApplication { The above example is somewhat of a the opposite from the previous sample and demonstrates function which takes single input of type `Integer` and produces two outputs (both of type `String`). -So, for the above example the input binding is `gather_in_0` and the -output bindings are `gather_out_0` and `gather_out_1`. +So, for the above example the input binding is `gather.in.0` and the +output bindings are `gather.out.0` and `gather.out.1`. And you test it with the following code: [source,java] @@ -664,11 +664,11 @@ As with functions with multiple inputs/outputs we can no longer rely on the nami destination bindings used by functions with single inputs/outputs. So we follow the same convention as for functions with multiple inputs/outputs: -* input - ` + _in_ + ` -* output - ` + _out_ + ` +* input - ` + .in. + ` +* output - ` + .out. + ` This means that the above configuration will result in the following destination bindings: -`uppercase_in_0`, `uppercase_out_0`, `reverse_in_0` and `reverse_out_0`. +`uppercase.in.0`, `uppercase.out.0`, `reverse.in.0` and `reverse.out.0`. And you test it with the following code: [source,java] diff --git a/pom.xml b/pom.xml index c3642ea6b..4092c24de 100644 --- a/pom.xml +++ b/pom.xml @@ -9,7 +9,7 @@ org.springframework.cloud spring-cloud-build - 2.2.0.BUILD-SNAPSHOT + 2.2.0.M5 @@ -23,16 +23,21 @@ 1.8 - Californium-SR8 + Californium-SR11 2.1 - 3.0.0.BUILD-SNAPSHOT - + 3.0.0.M3 true true true + + org.springframework.boot + spring-boot-actuator + 2.2.0.M6 + + org.springframework.cloud spring-cloud-function-context diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/CompositeMessageChannelConfigurer.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/CompositeMessageChannelConfigurer.java index 3cf40be90..b2edbfebe 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/CompositeMessageChannelConfigurer.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/CompositeMessageChannelConfigurer.java @@ -39,15 +39,19 @@ public class CompositeMessageChannelConfigurer @Override public void configureInputChannel(MessageChannel messageChannel, String channelName) { for (MessageChannelConfigurer messageChannelConfigurer : this.messageChannelConfigurers) { - messageChannelConfigurer.configureInputChannel(messageChannel, channelName); + if (messageChannelConfigurer != null) { + messageChannelConfigurer.configureInputChannel(messageChannel, channelName); + } + } } @Override - public void configureOutputChannel(MessageChannel messageChannel, - String channelName) { + public void configureOutputChannel(MessageChannel messageChannel, String channelName) { for (MessageChannelConfigurer messageChannelConfigurer : this.messageChannelConfigurers) { - messageChannelConfigurer.configureOutputChannel(messageChannel, channelName); + if (messageChannelConfigurer != null) { + messageChannelConfigurer.configureOutputChannel(messageChannel, channelName); + } } } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderFactoryAutoConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderFactoryAutoConfiguration.java index c9826d542..8e4ea78a4 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderFactoryAutoConfiguration.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderFactoryAutoConfiguration.java @@ -49,6 +49,7 @@ import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Import; import org.springframework.context.annotation.Role; +import org.springframework.core.env.Environment; import org.springframework.core.io.Resource; import org.springframework.core.io.UrlResource; import org.springframework.core.io.support.PropertiesLoaderUtils; @@ -199,7 +200,11 @@ public class BinderFactoryAutoConfiguration { @Bean public MessageConverterConfigurer messageConverterConfigurer( BindingServiceProperties bindingServiceProperties, - @Qualifier(IntegrationContextUtils.ARGUMENT_RESOLVER_MESSAGE_CONVERTER_BEAN_NAME) CompositeMessageConverter compositeMessageConverter) { + @Qualifier(IntegrationContextUtils.ARGUMENT_RESOLVER_MESSAGE_CONVERTER_BEAN_NAME) CompositeMessageConverter compositeMessageConverter, + Environment environment) { + if (StringUtils.hasText(environment.getProperty("spring.cloud.stream.function.definition"))) { + return null; + } return new MessageConverterConfigurer(bindingServiceProperties, compositeMessageConverter); } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/BindableFunctionProxyFactory.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/BindableFunctionProxyFactory.java index 600f0d4a9..dbc8e6ce1 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/BindableFunctionProxyFactory.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/BindableFunctionProxyFactory.java @@ -30,9 +30,9 @@ import org.springframework.util.CollectionUtils; * operates on Bindable interfaces (e.g., Source, Processor, Sink) which internally * define inputs and output channels. Unlike BindableProxyFactory, this class * operates based on the count of provided inputs and outputs deriving the binding - * (channel) names based on convention - {@code `_ + + _`} + * (channel) names based on convention - {@code `. + + .`} *
- * For example, `myFunction_in_0` - is the binding for the first input argument of the + * For example, `myFunction.in.0` - is the binding for the first input argument of the * function with the name `myFunction`. * * @author Oleg Zhurakousky @@ -41,6 +41,8 @@ import org.springframework.util.CollectionUtils; */ class BindableFunctionProxyFactory extends BindableProxyFactory { + static final String delimiter = "."; + private final int inputCount; private final int outputCount; @@ -125,11 +127,21 @@ class BindableFunctionProxyFactory extends BindableProxyFactory { } private String buildInputNameForIndex(int index) { - return this.functionDefinition + "_in_" + index; + return new StringBuilder(this.functionDefinition) + .append(delimiter) + .append("in") + .append(delimiter) + .append(index) + .toString(); } private String buildOutputNameForIndex(int index) { - return this.functionDefinition + "_out_" + index; + return new StringBuilder(this.functionDefinition) + .append(delimiter) + .append("out") + .append(delimiter) + .append(index) + .toString(); } private void createInput(String name) { 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 7f30eab53..83866ff7b 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 @@ -16,6 +16,7 @@ package org.springframework.cloud.stream.function; +import java.lang.reflect.Field; import java.lang.reflect.Method; import java.lang.reflect.Type; import java.time.Duration; @@ -53,8 +54,10 @@ import org.springframework.cloud.stream.annotation.EnableBinding; import org.springframework.cloud.stream.binder.BinderTypeRegistry; import org.springframework.cloud.stream.binder.BindingCreatedEvent; import org.springframework.cloud.stream.binder.ConsumerProperties; +import org.springframework.cloud.stream.binder.ProducerProperties; import org.springframework.cloud.stream.binding.BindableProxyFactory; import org.springframework.cloud.stream.config.BinderFactoryAutoConfiguration; +import org.springframework.cloud.stream.config.BindingBeansRegistrar; import org.springframework.cloud.stream.config.BindingProperties; import org.springframework.cloud.stream.config.BindingServiceConfiguration; import org.springframework.cloud.stream.config.BindingServiceProperties; @@ -72,6 +75,7 @@ import org.springframework.context.support.GenericApplicationContext; import org.springframework.core.annotation.AnnotationUtils; import org.springframework.core.env.Environment; import org.springframework.core.type.MethodMetadata; +import org.springframework.integration.channel.AbstractMessageChannel; import org.springframework.integration.channel.MessageChannelReactiveUtils; import org.springframework.integration.context.IntegrationObjectSupport; import org.springframework.integration.dsl.IntegrationFlow; @@ -82,6 +86,7 @@ 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.ClassUtils; @@ -97,8 +102,8 @@ import org.springframework.util.StringUtils; */ @Configuration @EnableConfigurationProperties(StreamFunctionProperties.class) -@Import(BinderFactoryAutoConfiguration.class) @AutoConfigureBefore(BindingServiceConfiguration.class) +@Import({ BindingBeansRegistrar.class, BinderFactoryAutoConfiguration.class }) public class FunctionConfiguration { @Bean @@ -124,7 +129,7 @@ public class FunctionConfiguration { */ @Bean IntegrationFlow supplierInitializer(FunctionCatalog functionCatalog, FunctionInspector functionInspector, - StreamFunctionProperties functionProperties, GenericApplicationContext context) { + StreamFunctionProperties functionProperties, GenericApplicationContext context, BindingServiceProperties serviceProperties) { if (!ObjectUtils.isEmpty(context.getBeanNamesForAnnotation(EnableBinding.class))) { return null; } @@ -135,7 +140,8 @@ public class FunctionConfiguration { : new String[] {}; for (String functionDefinition : functionDefinitions) { - FunctionInvocationWrapper functionWrapper = functionCatalog.lookup(functionDefinition); + String contentType = serviceProperties.getBindingProperties("output").getContentType(); + FunctionInvocationWrapper functionWrapper = functionCatalog.lookup(functionDefinition, contentType); if (functionWrapper != null && functionWrapper.isSupplier()) { Publisher beginPublishingTrigger = this.setupBindingTrigger(context); @@ -187,8 +193,7 @@ public class FunctionConfiguration { FunctionInspector inspector, Publisher beginPublishingTrigger, PollableSupplier pollable) { IntegrationFlowBuilder integrationFlowBuilder; - Type functionType = FunctionTypeUtils.getFunctionType(supplier, inspector); - + Type functionType = ((FunctionInvocationWrapper) supplier).getFunctionType(); boolean splittable = pollable != null && (boolean) AnnotationUtils.getAnnotationAttributes(pollable).get("splittable"); @@ -343,36 +348,37 @@ public class FunctionConfiguration { String channelType = (String) ((DirectWithAttributesChannel) messageChannel).getAttribute("type"); if (Source.OUTPUT.equals(channelType) && functionProperties.isComposeFrom()) { logger.info("Composing at the head of 'output' channel"); - BindingProperties properties = this.serviceProperties.getBindings().get(channelName); + BindingProperties properties = this.serviceProperties.getBindingProperties(channelName); FunctionInvocationWrapper function = functionCatalog.lookup(functionDefinition, properties.getContentType()); - this.composeSimpleFunctionToExistingFlow(function, messageChannel, channelName, bindableProxyFactory); + this.composeSimpleFunctionToExistingFlow(function, messageChannel, bindableProxyFactory); } else { - BindingProperties properties = this.serviceProperties.getBindings().get(channelName); + BindingProperties properties = this.serviceProperties.getBindingProperties(channelName); FunctionInvocationWrapper function = functionCatalog.lookup(functionDefinition, properties.getContentType()); this.bindSimpleFunctions(function, messageChannel, bindableProxyFactory); } } - private void composeSimpleFunctionToExistingFlow(FunctionInvocationWrapper function, SubscribableChannel messageChannel, - String channelName, BindableProxyFactory bindableProxyFactory) { - ServiceActivatingHandler handler = createFunctionHandler(function); + private void composeSimpleFunctionToExistingFlow(FunctionInvocationWrapper function, SubscribableChannel outputChannel, + BindableProxyFactory bindableProxyFactory) { + String outputChannelName = ((AbstractMessageChannel) outputChannel).getBeanName(); + ServiceActivatingHandler handler = createFunctionHandler(function, null, outputChannelName); DirectWithAttributesChannel newOutputChannel = new DirectWithAttributesChannel(); newOutputChannel.setAttribute("type", "output"); newOutputChannel.setComponentName("output.extended"); this.context.registerBean("output.extended", MessageChannel.class, () -> newOutputChannel); - bindableProxyFactory.replaceOutputChannel(channelName, "output.extended", newOutputChannel); + bindableProxyFactory.replaceOutputChannel(outputChannelName, "output.extended", newOutputChannel); handler.setOutputChannelName("output.extended"); - messageChannel.subscribe(handler); + outputChannel.subscribe(handler); } private void bindSimpleFunctions(FunctionInvocationWrapper function, SubscribableChannel inputChannel, BindableProxyFactory bindableProxyFactory) { - Type functionType = FunctionTypeUtils.getFunctionType(function, this.functionInspector); + Type functionType = function.getFunctionType(); String outputChannelName = bindableProxyFactory instanceof BindableFunctionProxyFactory ? ((BindableFunctionProxyFactory) bindableProxyFactory).getOutputName(0) - : Source.OUTPUT; + : (FunctionTypeUtils.isConsumer(functionType) ? null : "output"); if (FunctionTypeUtils.isReactive(FunctionTypeUtils.getInputType(functionType, 0))) { MessageChannel outputChannel = context.getBean(outputChannelName, MessageChannel.class); @@ -382,7 +388,8 @@ public class FunctionConfiguration { this.subscribeToInput(function, publisher, message -> outputChannel.send((Message) message)); } else { - ServiceActivatingHandler handler = createFunctionHandler(function); + String inputChannelName = ((AbstractMessageChannel) inputChannel).getBeanName(); + ServiceActivatingHandler handler = createFunctionHandler(function, inputChannelName, outputChannelName); if (!FunctionTypeUtils.isConsumer(functionType)) { handler.setOutputChannelName(outputChannelName); } @@ -390,8 +397,15 @@ public class FunctionConfiguration { } } - private ServiceActivatingHandler createFunctionHandler(FunctionInvocationWrapper function) { - ServiceActivatingHandler handler = new ServiceActivatingHandler(new FunctionWrapper(function)); + private ServiceActivatingHandler createFunctionHandler(FunctionInvocationWrapper function, + String inputChannelName, String outputChannelName) { + ConsumerProperties consumerProperties = StringUtils.hasText(inputChannelName) + ? this.serviceProperties.getBindingProperties(inputChannelName).getConsumer() + : null; + ProducerProperties producerProperties = StringUtils.hasText(outputChannelName) + ? this.serviceProperties.getBindingProperties(outputChannelName).getProducer() + : null; + ServiceActivatingHandler handler = new ServiceActivatingHandler(new FunctionWrapper(function, consumerProperties, producerProperties)); handler.setBeanFactory(context); handler.afterPropertiesSet(); return handler; @@ -484,15 +498,31 @@ public class FunctionConfiguration { private static class FunctionWrapper implements Function, Object> { private final Function function; - FunctionWrapper(Function function) { + private final ConsumerProperties consumerProperties; + + private final ProducerProperties producerProperties; + + private final Field headersField; + + FunctionWrapper(Function function, ConsumerProperties consumerProperties, ProducerProperties producerProperties) { this.function = function; + this.consumerProperties = consumerProperties; + this.producerProperties = producerProperties; + this.headersField = ReflectionUtils.findField(MessageHeaders.class, "headers"); + this.headersField.setAccessible(true); } @SuppressWarnings("unchecked") @Override - public Message apply(Message t) { - Object result = function.apply(t); + public Message apply(Message message) { + +// Map headersMap = (Map) ReflectionUtils +// .getField(this.headersField, message.getHeaders()); + + + Object result = function.apply(message); if (result instanceof Publisher) { - throw new IllegalStateException("Routing to functions that return Publisher is not supported in the context of Spring Cloud Stream."); + throw new IllegalStateException("Routing to functions that return Publisher is not supported " + + "in the context of Spring Cloud Stream."); } return (Message) result; } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java index 0f3d83b72..32397b972 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java @@ -16,8 +16,10 @@ package org.springframework.cloud.stream.function; +import java.io.Serializable; import java.util.function.Consumer; import java.util.function.Function; +import java.util.function.Supplier; import org.junit.After; import org.junit.Test; @@ -61,7 +63,7 @@ public class ImplicitFunctionBindingTests { TestChannelBinderConfiguration.getCompleteConfiguration( EmptyConfiguration.class)) .web(WebApplicationType.NONE) - .run("--spring.jmx.enabled=false")) { + .run("--spring.jmx.enabled=false", "--debug")) { context.getBean(InputDestination.class); } catch (Exception e) { // should not fail @@ -166,7 +168,9 @@ public class ImplicitFunctionBindingTests { .getCompleteConfiguration(SingleConsumerConfiguration.class)) .web(WebApplicationType.NONE) .run("--spring.cloud.stream.function.definition=consumer", - "--spring.jmx.enabled=false")) { + "--spring.jmx.enabled=false", + "--spring.cloud.stream.bindings.input.content-type=text/plain", + "--spring.cloud.stream.bindings.input.consumer.use-native-decoding=true")) { InputDestination source = context.getBean(InputDestination.class); source.send(new GenericMessage("John Doe".getBytes())); @@ -214,6 +218,24 @@ public class ImplicitFunctionBindingTests { } + @Test(expected = Exception.class) + public void testDeclaredTypeVsActualInstance() { + System.clearProperty("spring.cloud.stream.function.definition"); + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration( + SCF_GH_409Configuration.class)) + .web(WebApplicationType.NONE) + .run("--spring.jmx.enabled=false")) { + + InputDestination inputDestination = context.getBean(InputDestination.class); + + Message inputMessageOne = MessageBuilder + .withPayload("Hello".getBytes()).build(); + + inputDestination.send(inputMessageOne); + } + } + @Test public void testWithContextTypeApplicationProperty() { System.clearProperty("spring.cloud.stream.function.definition"); @@ -312,4 +334,23 @@ public class ImplicitFunctionBindingTests { } + @EnableAutoConfiguration + public static class SCF_GH_409Configuration { + + @Bean + public Serializable blah() { + return new Foo(); + } + + private static class Foo implements Supplier, Serializable { + + @Override + public Object get() { + // TODO Auto-generated method stub + return null; + } + + } + } + }