SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Changed '_' to '.' for convention used for function bean name delimiter

Browse Source 
Disconneced MessageChannelConfigurer from function processing

Upgraded to SCF M3 and Californium SR11
This commit is contained in:
Oleg Zhurakousky
2019-09-21 15:05:56 +02:00
parent c56b4d1ea9
commit 19f843ed18
7 changed files with 145 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
docs/src/main/asciidoc/spring-cloud-stream.adoc
View File

@@ -542,8 +542,8 @@ So, this is where understanding of the naming convention for binding destinatio
*Binding naming convention:*
* input - `<functionName> + _in_ + <index>`
* output - `<functionName> + _out_ + <index>`
* input - `<functionName> + .in. + <index>`
* output - `<functionName> + .out. + <index>`
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 - `<functionName> + _in_ + <index>`
* output - `<functionName> + _out_ + <index>`
* input - `<functionName> + .in. + <index>`
* output - `<functionName> + .out. + <index>`
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]

13
pom.xml
View File

@@ -9,7 +9,7 @@
<parent>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-build</artifactId>
<version>2.2.0.BUILD-SNAPSHOT</version>
<version>2.2.0.M5</version>
<relativePath/>
</parent>
<scm>
@@ -23,16 +23,21 @@
</scm>
<properties>
<java.version>1.8</java.version>
<reactor.version>Californium-SR8</reactor.version>
<reactor.version>Californium-SR11</reactor.version>
<objenesis.version>2.1</objenesis.version>
<spring-cloud-function.version>3.0.0.BUILD-SNAPSHOT</spring-cloud-function.version>
<spring-cloud-function.version>3.0.0.M3</spring-cloud-function.version>
<maven-checkstyle-plugin.failsOnError>true</maven-checkstyle-plugin.failsOnError>
<maven-checkstyle-plugin.failsOnViolation>true</maven-checkstyle-plugin.failsOnViolation>
<maven-checkstyle-plugin.includeTestSourceDirectory>true</maven-checkstyle-plugin.includeTestSourceDirectory>
</properties>
<dependencyManagement>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-actuator</artifactId>
<version>2.2.0.M6</version>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-function-context</artifactId>

12
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/CompositeMessageChannelConfigurer.java
View File

@@ -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);
}
}
}

7
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderFactoryAutoConfiguration.java
View File

@@ -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);
}

20
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/BindableFunctionProxyFactory.java
View File

@@ -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 `<function-definition>_ + <in/out> + _<index>`}
* (channel) names based on convention - {@code `<function-definition>. + <in/out> + .<index>`}
* <br>
* 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) {

74
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java
View File

@@ -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<Object> beginPublishingTrigger = this.setupBindingTrigger(context);
@@ -187,8 +193,7 @@ public class FunctionConfiguration {
FunctionInspector inspector, Publisher<Object> 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<Message<byte[]>, 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<byte[]> apply(Message<byte[]> t) {
Object result = function.apply(t);
public Message<byte[]> apply(Message<byte[]> message) {
// Map<String, Object> headersMap = (Map<String, Object>) 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<byte[]>) result;
}

45
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java
View File

@@ -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<byte[]>("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<byte[]> 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<Object>, Serializable {
@Override
public Object get() {
// TODO Auto-generated method stub
return null;
}
}
}
}