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

Add initial support for functions wih multiple inputs/outputs as well as multiple functions

Browse Source 
- Currently there is no multi-out support for Supplier and I am not really sure what the use case would be so holding off.
- The logic in FunctionConfiguration effectively split in bootstrapping simple functions (e.g., Function<String, String>) vs. multi-in/out (e.g., Function<Tuple2<Flux<String>, Flux<String>>, Flux<String>>). This is temporary given that multi-in/out is still considered WIP. Once it becomes stable we can merge the two two for consistency.
- Added multiple input/output support for TestBinder
- Updated documentation

Resolves #1746
Resolves #1745
Resolves #1314
This commit is contained in:
Oleg Zhurakousky
2019-09-16 11:51:21 +02:00
parent 1723a40786
commit c56b4d1ea9
10 changed files with 1169 additions and 1045 deletions
Download Patch File Download Diff File Expand all files Collapse all files

41
README.adoc
View File

@@ -105,16 +105,17 @@ Modify the `com.example.loggingconsumer.LoggingConsumerApplication` class to loo
[source, java]
----
@SpringBootApplication
@EnableBinding(Sink.class)
public class LoggingConsumerApplication {
public static void main(String[] args) {
SpringApplication.run(LoggingConsumerApplication.class, args);
}
@StreamListener(Sink.INPUT)
public void handle(Person person) {
System.out.println("Received: " + person);
@Bean
public Consumer<Person> log() {
return person -> {
System.out.println("Received: " + person);
};
}
public static class Person {
@@ -134,10 +135,10 @@ public class LoggingConsumerApplication {
As you can see from the preceding listing:
* We have enabled `Sink` binding (input-no-output) by using `@EnableBinding(Sink.class)`.
Doing so signals to the framework to initiate binding to the messaging middleware, where it automatically creates the destination (that is, queue, topic, and others) that are bound to the `Sink.INPUT` channel.
* We have added a `handler` method to receive incoming messages of type `Person`.
Doing so lets you see one of the core features of the framework: It tries to automatically convert incoming message payloads to type `Person`.
* We are using functional programming model (see <<Spring Cloud Function support>>) to define a single message handler as `Consumer`.
* We are relying on framework conventions to bind such handler to the input destination binding exposed by the binder.
Doing so also lets you see one of the core features of the framework: It tries to automatically convert incoming message payloads to type `Person`.
You now have a fully functional Spring Cloud Stream application that does listens for messages.
From here, for simplicity, we assume you selected RabbitMQ in <<spring-cloud-stream-preface-creating-sample-application,step one>>.
@@ -172,35 +173,21 @@ You can also build and package your application into a boot jar (by using `./mvn
Now you have a working (albeit very basic) Spring Cloud Stream application.
== What's New in 2.2?
Spring Cloud Stream introduces a number of new features, enhancements, and changes in addition to the once already introduced in
https://docs.spring.io/spring-cloud-stream/docs/Elmhurst.SR2/reference/htmlsingle/#_what_s_new_in_2_0[version 2.0]
The following sections outline the most notable ones:
* <<spring-cloud-stream-preface-new-features>>
* <<spring-cloud-stream-preface-notable-enhancements>>
== What's New in 3.0?
TBD
[[spring-cloud-stream-preface-new-features]]
=== New Features and Components
TBD
[[spring-cloud-stream-preface-notable-enhancements]]
=== Notable Enhancements
TBD
[[spring-cloud-stream-preface-notable-deprecations]]
=== Notable Deprecations
As of version 2.2, the following items have been deprecated:
- The spring-cloud-stream-reactive module is deprecated in favor of native support
via <<spring_cloud_function, Spring Cloud Function>> programming model.
=== Notes on migrating from 1.x to 2.x?
- Due to the improvements in content-type negotiation, the `originalContentType` header is not used (ignored) since 2.x and only exists for maintaining compatibility with 1.x versions
- Introduction of `@StreamRetryTemplate` qualifier. While configuring custom instance of the `RetryTemplate` and to avoid conflicts you must qualify the instance of such `RetryTemplate` with this qualifier. See <<Retry Template, Retry Template>> for more details.
TBD
= Appendices
[appendix]

41
docs/src/main/asciidoc/preface.adoc
View File

@@ -85,16 +85,17 @@ Modify the `com.example.loggingconsumer.LoggingConsumerApplication` class to loo
[source, java]
----
@SpringBootApplication
@EnableBinding(Sink.class)
public class LoggingConsumerApplication {
public static void main(String[] args) {
SpringApplication.run(LoggingConsumerApplication.class, args);
}
@StreamListener(Sink.INPUT)
public void handle(Person person) {
System.out.println("Received: " + person);
@Bean
public Consumer<Person> log() {
return person -> {
System.out.println("Received: " + person);
};
}
public static class Person {
@@ -114,10 +115,10 @@ public class LoggingConsumerApplication {
As you can see from the preceding listing:
* We have enabled `Sink` binding (input-no-output) by using `@EnableBinding(Sink.class)`.
Doing so signals to the framework to initiate binding to the messaging middleware, where it automatically creates the destination (that is, queue, topic, and others) that are bound to the `Sink.INPUT` channel.
* We have added a `handler` method to receive incoming messages of type `Person`.
Doing so lets you see one of the core features of the framework: It tries to automatically convert incoming message payloads to type `Person`.
* We are using functional programming model (see <<Spring Cloud Function support>>) to define a single message handler as `Consumer`.
* We are relying on framework conventions to bind such handler to the input destination binding exposed by the binder.
Doing so also lets you see one of the core features of the framework: It tries to automatically convert incoming message payloads to type `Person`.
You now have a fully functional Spring Cloud Stream application that does listens for messages.
From here, for simplicity, we assume you selected RabbitMQ in <<spring-cloud-stream-preface-creating-sample-application,step one>>.
@@ -152,32 +153,18 @@ You can also build and package your application into a boot jar (by using `./mvn
Now you have a working (albeit very basic) Spring Cloud Stream application.
== What's New in 2.2?
Spring Cloud Stream introduces a number of new features, enhancements, and changes in addition to the once already introduced in
https://docs.spring.io/spring-cloud-stream/docs/Elmhurst.SR2/reference/htmlsingle/#_what_s_new_in_2_0[version 2.0]
The following sections outline the most notable ones:
* <<spring-cloud-stream-preface-new-features>>
* <<spring-cloud-stream-preface-notable-enhancements>>
== What's New in 3.0?
TBD
[[spring-cloud-stream-preface-new-features]]
=== New Features and Components
TBD
[[spring-cloud-stream-preface-notable-enhancements]]
=== Notable Enhancements
TBD
[[spring-cloud-stream-preface-notable-deprecations]]
=== Notable Deprecations
As of version 2.2, the following items have been deprecated:
- The spring-cloud-stream-reactive module is deprecated in favor of native support
via <<spring_cloud_function, Spring Cloud Function>> programming model.
=== Notes on migrating from 1.x to 2.x?
- Due to the improvements in content-type negotiation, the `originalContentType` header is not used (ignored) since 2.x and only exists for maintaining compatibility with 1.x versions
- Introduction of `@StreamRetryTemplate` qualifier. While configuring custom instance of the `RetryTemplate` and to avoid conflicts you must qualify the instance of such `RetryTemplate` with this qualifier. See <<Retry Template, Retry Template>> for more details.
TBD

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

File diff suppressed because it is too large Load Diff

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

@@ -22,31 +22,41 @@ import org.springframework.cloud.stream.binding.BoundTargetHolder;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
/**
* {@link FactoryBean} for creating inputs/outputs destinations to be bound to
* function arguments. It is an extension to {@link BindableProxyFactory} which
* operates on Bindable interfaces (e.g., Source, Processor, Sink) which internally
* define inputs and output channels. Unlike BindableProxyFactory, this class simply
* operates based on the count of provided inputs and outputs and the names of inputs and outputs
* are based on convention.
* 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>`}
* <br>
* For example, `myFunction_in_0` - is the binding for the first input argument of the
* function with the name `myFunction`.
*
* @author Oleg Zhurakousky
*
*
*
* @since 3.0
*/
public class BindableFunctionProxyFactory extends BindableProxyFactory {
class BindableFunctionProxyFactory extends BindableProxyFactory {
private final int inputCount;
private final int outputCount;
public BindableFunctionProxyFactory(int inputCount, int outputCount) {
private final String functionDefinition;
private final boolean nameBasedOnFunctionName;
private boolean multiple;
BindableFunctionProxyFactory(String functionDefinition, int inputCount, int outputCount, boolean nameBasedOnFunctionName) {
super(null);
this.inputCount = inputCount;
this.outputCount = outputCount;
this.functionDefinition = functionDefinition;
this.nameBasedOnFunctionName = nameBasedOnFunctionName;
}
@@ -55,25 +65,73 @@ public class BindableFunctionProxyFactory extends BindableProxyFactory {
Assert.notEmpty(BindableFunctionProxyFactory.this.bindingTargetFactories,
"'bindingTargetFactories' cannot be empty");
this.multiple = this.inputCount > 1 || this.outputCount > 1;
if (this.inputCount > 0) {
if (this.inputCount == 1) {
this.createInput("input");
if (multiple || nameBasedOnFunctionName) {
for (int i = 0; i < inputCount; i++) {
this.createInput(this.buildInputNameForIndex(i));
}
}
else {
throw new UnsupportedOperationException("Multiple inputs are not currently supported");
this.createInput("input");
}
}
if (this.outputCount > 0) {
if (this.outputCount == 1) {
this.createOutput("output");
if (multiple || nameBasedOnFunctionName) {
for (int i = 0; i < outputCount; i++) {
this.createOutput(this.buildOutputNameForIndex(i));
}
}
else {
throw new UnsupportedOperationException("Multiple outputs are not currently supported");
this.createOutput("output");
}
}
}
@Override
public Class<?> getObjectType() {
return this.type;
}
@Override
public boolean isSingleton() {
return true;
}
protected boolean isNameBasedOnFunctionName() {
return this.nameBasedOnFunctionName;
}
protected String getFunctionDefinition() {
return this.functionDefinition;
}
protected String getInputName(int index) {
return CollectionUtils.isEmpty(this.getInputs())
? null
: this.getInputs().toArray(new String[0])[index];
}
protected String getOutputName(int index) {
return CollectionUtils.isEmpty(this.getOutputs())
? null
: this.getOutputs().toArray(new String[0])[index];
}
protected boolean isMultiple() {
return this.multiple;
}
private String buildInputNameForIndex(int index) {
return this.functionDefinition + "_in_" + index;
}
private String buildOutputNameForIndex(int index) {
return this.functionDefinition + "_out_" + index;
}
private void createInput(String name) {
BindableFunctionProxyFactory.this.inputHolders.put(name,
new BoundTargetHolder(getBindingTargetFactory(SubscribableChannel.class)
@@ -86,15 +144,4 @@ public class BindableFunctionProxyFactory extends BindableProxyFactory {
.createOutput(name), true));
}
@Override
public Class<?> getObjectType() {
return this.type;
}
@Override
public boolean isSingleton() {
return true;
}
}

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

@@ -19,6 +19,8 @@ package org.springframework.cloud.stream.function;
import java.lang.reflect.Method;
import java.lang.reflect.Type;
import java.time.Duration;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
import java.util.function.Function;
@@ -31,6 +33,7 @@ import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.MonoSink;
import reactor.util.function.Tuples;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.InitializingBean;
@@ -70,6 +73,7 @@ import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.core.env.Environment;
import org.springframework.core.type.MethodMetadata;
import org.springframework.integration.channel.MessageChannelReactiveUtils;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlowBuilder;
import org.springframework.integration.dsl.IntegrationFlows;
@@ -78,11 +82,9 @@ 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;
import org.springframework.util.MimeTypeUtils;
import org.springframework.util.ObjectUtils;
import org.springframework.util.ReflectionUtils;
import org.springframework.util.StringUtils;
@@ -99,83 +101,87 @@ import org.springframework.util.StringUtils;
@AutoConfigureBefore(BindingServiceConfiguration.class)
public class FunctionConfiguration {
/*
* Creates an effective representation of Bindable interfaces by maintaining the count of inputs and
* outputs based on the provided function, thus preserving the contract and the infrastructure code used
* by current EnableBinding/StreamListener combination.
* It is then used buy `functionInitializer` or 'supplierInitializer` where functions are actually bound to channels.
*
* Also, see the BindableFunctionProxyFactory
*/
@Bean
public InitializingBean functionBindingHolder(Environment environment, FunctionCatalog functionCatalog,
public InitializingBean functionBindingRegistrar(Environment environment, FunctionCatalog functionCatalog,
StreamFunctionProperties streamFunctionProperties, BinderTypeRegistry binderTypeRegistry) {
return new FunctionBindingHolder(binderTypeRegistry, functionCatalog, streamFunctionProperties);
return new FunctionBindingRegistrar(binderTypeRegistry, functionCatalog, streamFunctionProperties);
}
@Bean
public InitializingBean functionInitializer(FunctionCatalog functionCatalog, FunctionInspector functionInspector,
StreamFunctionProperties functionProperties, @Nullable BindableProxyFactory[] bpfs, BindingServiceProperties serviceProperties,
ConfigurableApplicationContext applicationContext, FunctionBindingHolder bindingHolder) {
StreamFunctionProperties functionProperties, @Nullable BindableProxyFactory[] bindableProxyFactories,
BindingServiceProperties serviceProperties, ConfigurableApplicationContext applicationContext, FunctionBindingRegistrar bindingHolder) {
if (bpfs == null || bpfs.length > 1) {
return null; // basically we're not dealing with multiple EnableBinding which is how multiple BindableProxyFactory are created
}
BindableProxyFactory bindableProxyFactory = bpfs[0];
return bindingHolder.getInputCount() > 0 // basically not a Supplier
|| !ObjectUtils.isEmpty(applicationContext.getBeanNamesForAnnotation(EnableBinding.class)) // implies existing binding to which we are going to 'compose to'
? new FunctionChannelBindingInitializer(functionCatalog, functionInspector, functionProperties, bindableProxyFactory, serviceProperties)
: null;
}
@Bean
public IntegrationFlow supplierInitializer(FunctionCatalog functionCatalog, FunctionInspector functionInspector,
StreamFunctionProperties functionProperties, GenericApplicationContext context, FunctionBindingHolder bindingHolder) {
if (bindingHolder.getInputCount() > 0) {
if (bindableProxyFactories == null) {
return null;
}
FunctionInvocationWrapper functionWrapper = functionCatalog.lookup(functionProperties.getDefinition());
IntegrationFlow integrationFlow = null;
if (ObjectUtils.isEmpty(context.getBeanNamesForAnnotation(EnableBinding.class)) && functionWrapper != null && functionWrapper.isSupplier()) {
AtomicReference<MonoSink<Object>> triggerRef = new AtomicReference<>();
Publisher<Object> beginPublishingTrigger = Mono.create(emmiter -> {
triggerRef.set(emmiter);
});
context.addApplicationListener(event -> {
if (event instanceof BindingCreatedEvent) {
if (triggerRef.get() != null) {
triggerRef.get().success();
}
}
});
return new FunctionChannelBindingInitializer(functionCatalog, functionInspector, functionProperties, bindableProxyFactories, serviceProperties);
}
RootBeanDefinition bd = (RootBeanDefinition) context.getBeanDefinition(functionProperties.getParsedDefinition()[0]);
Method factoryMethod = bd.getResolvedFactoryMethod();
if (factoryMethod == null) {
Object source = bd.getSource();
if (source instanceof MethodMetadata) {
Class<?> factory = ClassUtils.resolveClassName(((MethodMetadata) source).getDeclaringClassName(), null);
Class<?>[] params = FunctionContextUtils.getParamTypesFromBeanDefinitionFactory(factory, bd);
factoryMethod = ReflectionUtils.findMethod(factory, ((MethodMetadata) source).getMethodName(), params);
}
}
Assert.notNull(factoryMethod, "Failed to introspect factory method since it was not discovered for function '"
+ functionProperties.getDefinition() + "'");
PollableSupplier pollable = factoryMethod.getReturnType().isAssignableFrom(Supplier.class)
? AnnotationUtils.findAnnotation(factoryMethod, PollableSupplier.class)
: null;
if (!functionProperties.isComposeFrom() && !functionProperties.isComposeTo()) {
integrationFlow = this.integrationFlowFromProvidedSupplier(functionWrapper, functionInspector, beginPublishingTrigger, pollable)
.channel("output").get();
}
/*
* Binding initializer responsible only for Suppliers only
*/
@Bean
IntegrationFlow supplierInitializer(FunctionCatalog functionCatalog, FunctionInspector functionInspector,
StreamFunctionProperties functionProperties, GenericApplicationContext context) {
if (!ObjectUtils.isEmpty(context.getBeanNamesForAnnotation(EnableBinding.class))) {
return null;
}
IntegrationFlow integrationFlow = null;
String[] functionDefinitions = StringUtils.hasText(functionProperties.getDefinition())
? functionProperties.getDefinition().split(";")
: new String[] {};
for (String functionDefinition : functionDefinitions) {
FunctionInvocationWrapper functionWrapper = functionCatalog.lookup(functionDefinition);
if (functionWrapper != null && functionWrapper.isSupplier()) {
Publisher<Object> beginPublishingTrigger = this.setupBindingTrigger(context);
RootBeanDefinition bd = (RootBeanDefinition) context.getBeanDefinition(functionProperties.getParsedDefinition()[0]);
Method factoryMethod = bd.getResolvedFactoryMethod();
if (factoryMethod == null) {
Object source = bd.getSource();
if (source instanceof MethodMetadata) {
Class<?> factory = ClassUtils.resolveClassName(((MethodMetadata) source).getDeclaringClassName(), null);
Class<?>[] params = FunctionContextUtils.getParamTypesFromBeanDefinitionFactory(factory, bd);
factoryMethod = ReflectionUtils.findMethod(factory, ((MethodMetadata) source).getMethodName(), params);
}
}
Assert.notNull(factoryMethod, "Failed to introspect factory method since it was not discovered for function '"
+ functionProperties.getDefinition() + "'");
PollableSupplier pollable = factoryMethod.getReturnType().isAssignableFrom(Supplier.class)
? AnnotationUtils.findAnnotation(factoryMethod, PollableSupplier.class)
: null;
if (!functionProperties.isComposeFrom() && !functionProperties.isComposeTo()) {
integrationFlow = this.integrationFlowFromProvidedSupplier(functionWrapper, functionInspector, beginPublishingTrigger, pollable)
.channel("output").get();
}
}
}
return integrationFlow;
}
/*
* Creates a publishing trigger to ensure Supplier does not begin publishing until binding is created
*/
private Publisher<Object> setupBindingTrigger(GenericApplicationContext context) {
AtomicReference<MonoSink<Object>> triggerRef = new AtomicReference<>();
Publisher<Object> beginPublishingTrigger = Mono.create(emmiter -> {
triggerRef.set(emmiter);
});
context.addApplicationListener(event -> {
if (event instanceof BindingCreatedEvent) {
if (triggerRef.get() != null) {
triggerRef.get().success();
}
}
});
return beginPublishingTrigger;
}
@SuppressWarnings({ "rawtypes", "unchecked" })
private IntegrationFlowBuilder integrationFlowFromProvidedSupplier(Supplier<?> supplier,
FunctionInspector inspector, Publisher<Object> beginPublishingTrigger, PollableSupplier pollable) {
@@ -206,13 +212,13 @@ public class FunctionConfiguration {
@SuppressWarnings("unchecked")
private <T> Message<T> wrapToMessageIfNecessary(T value) {
return value instanceof Message ? (Message<T>) value : MessageBuilder.withPayload(value).setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON).build();
return value instanceof Message
? (Message<T>) value
: MessageBuilder.withPayload(value).build();
}
/**
*
* @author Oleg Zhurakousky
* @since 3.0
/*
* Binding initializer responsible only for Functions and Consumers.
*/
private static class FunctionChannelBindingInitializer implements InitializingBean, ApplicationContextAware {
@@ -224,7 +230,7 @@ public class FunctionConfiguration {
private final StreamFunctionProperties functionProperties;
private final BindableProxyFactory bindableProxyFactory;
private final BindableProxyFactory[] bindableProxyFactories;
private final BindingServiceProperties serviceProperties;
@@ -232,35 +238,38 @@ public class FunctionConfiguration {
FunctionChannelBindingInitializer(FunctionCatalog functionCatalog, FunctionInspector functionInspector,
StreamFunctionProperties functionProperties, BindableProxyFactory bindableProxyFactory, BindingServiceProperties serviceProperties) {
StreamFunctionProperties functionProperties, BindableProxyFactory[] bindableProxyFactories, BindingServiceProperties serviceProperties) {
this.functionCatalog = functionCatalog;
this.functionInspector = functionInspector;
this.functionProperties = functionProperties;
this.bindableProxyFactory = bindableProxyFactory;
this.bindableProxyFactories = bindableProxyFactories;
this.serviceProperties = serviceProperties;
}
@Override
public void afterPropertiesSet() throws Exception {
MessageChannel messageChannel = null;
String channelName = Sink.INPUT;
if (context.containsBean(channelName)) {
Object bean = context.getBean(channelName);
if (bean instanceof MessageChannel) {
messageChannel = context.getBean(channelName, MessageChannel.class);
Stream.of(this.bindableProxyFactories).forEach(bindableProxyFactory -> {
String functionDefinition = getFunctionDefinition(bindableProxyFactory);
FunctionInvocationWrapper function = functionCatalog.lookup(functionDefinition);
if (function != null && !function.isSupplier()) {
if (isMultipleInputOutput(bindableProxyFactory)) {
this.bindMultipleArgumentsFunction(bindableProxyFactory, functionDefinition);
}
else {
SubscribableChannel messageChannel = this.determineChannelToSubscribeTo(bindableProxyFactory);
if (messageChannel != null && function != null) {
this.bindOrComposeSimpleFunctions(((IntegrationObjectSupport) messageChannel).getComponentName(),
(SubscribableChannel) messageChannel, bindableProxyFactory, functionDefinition);
}
}
}
}
if (messageChannel == null && context.containsBean(Source.OUTPUT)) {
channelName = "output";
Object bean = context.getBean(channelName);
if (bean instanceof MessageChannel) {
messageChannel = context.getBean(channelName, SubscribableChannel.class);
}
}
});
}
if (messageChannel != null && functionCatalog.lookup(functionProperties.getDefinition()) != null) {
this.doPostProcess(channelName, (SubscribableChannel) messageChannel);
}
private String getFunctionDefinition(BindableProxyFactory bindableProxyFactory) {
return bindableProxyFactory instanceof BindableFunctionProxyFactory
? ((BindableFunctionProxyFactory) bindableProxyFactory).getFunctionDefinition()
: functionProperties.getDefinition();
}
@Override
@@ -268,67 +277,134 @@ public class FunctionConfiguration {
this.context = (GenericApplicationContext) applicationContext;
}
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 && Sink.INPUT.equals(channelName)) {
throw new UnsupportedOperationException("Composing at tail is not currently supported");
}
else if (functionProperties.isComposeFrom() && Source.OUTPUT.equals(channelName)) {
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");
BindingProperties properties = this.serviceProperties.getBindings().get(Source.OUTPUT);
FunctionInvocationWrapper function = functionCatalog.lookup(functionProperties.getDefinition(), properties.getContentType());
ServiceActivatingHandler handler = new ServiceActivatingHandler(new FunctionWrapper(function));
handler.setBeanFactory(context);
handler.afterPropertiesSet();
DirectWithAttributesChannel newOutputChannel = new DirectWithAttributesChannel();
newOutputChannel.setAttribute("type", "output");
newOutputChannel.setComponentName("output.extended");
this.context.registerBean("output.extended", MessageChannel.class, () -> newOutputChannel);
this.bindableProxyFactory.replaceOutputChannel(channelName, "output.extended", newOutputChannel);
handler.setOutputChannelName("output.extended");
SubscribableChannel subscribeChannel = (SubscribableChannel) messageChannel;
subscribeChannel.subscribe(handler);
private SubscribableChannel determineChannelToSubscribeTo(BindableProxyFactory bindableProxyFactory) {
SubscribableChannel messageChannel = null;
if (bindableProxyFactory instanceof BindableFunctionProxyFactory) {
String channelName = ((BindableFunctionProxyFactory) bindableProxyFactory).getInputName(0);
messageChannel = context.getBean(channelName, SubscribableChannel.class);
}
else {
if (Sink.INPUT.equals(channelName)) {
BindingProperties properties = this.serviceProperties.getBindings().get(Sink.INPUT);
FunctionInvocationWrapper function = functionCatalog.lookup(functionProperties.getDefinition(), properties.getContentType());
this.postProcessForStandAloneFunction(function, messageChannel);
// could be "input" or "output" if subscribing to existing Source
if (context.containsBean(Sink.INPUT)) {
messageChannel = context.getBean(Sink.INPUT, SubscribableChannel.class);
}
else if (context.containsBean(Source.OUTPUT)) {
messageChannel = context.getBean(Source.OUTPUT, SubscribableChannel.class);
}
}
return messageChannel;
}
@SuppressWarnings({ "unchecked", "rawtypes" })
private void bindMultipleArgumentsFunction(BindableProxyFactory bindableProxyFactory, String functionDefinition) {
Assert.isTrue(!functionProperties.isComposeTo() && !functionProperties.isComposeFrom(),
"Composing to/from existing Sinks and Sources are not supported for functions with multiple arguments.");
BindableFunctionProxyFactory functionProxyFactory = (BindableFunctionProxyFactory) bindableProxyFactory;
Set<String> inputBindingNames = functionProxyFactory.getInputs();
Set<String> outputBindingNames = functionProxyFactory.getOutputs();
String[] outputContentTypes = outputBindingNames.stream()
.map(bindingName -> this.serviceProperties.getBindings().get(bindingName).getContentType())
.toArray(String[]::new);
FunctionInvocationWrapper function = functionCatalog.lookup(functionDefinition, outputContentTypes);
if (isMultipleInputOutput(bindableProxyFactory)) {
this.assertSupportedSignatures(function.getFunctionType());
}
Publisher[] inputPublishers = inputBindingNames.stream().map(inputBindingName -> {
SubscribableChannel inputChannel = context.getBean(inputBindingName, SubscribableChannel.class);
return this.enhancePublisher(MessageChannelReactiveUtils.toPublisher(inputChannel), inputBindingName);
}).toArray(Publisher[]::new);
Object resultPublishers = function.apply(inputPublishers.length == 1 ? inputPublishers[0] : Tuples.fromArray(inputPublishers));
if (resultPublishers instanceof Iterable) {
Iterator<String> outputBindingIter = outputBindingNames.iterator();
((Iterable) resultPublishers).forEach(publisher -> {
MessageChannel outputChannel = context.getBean(outputBindingIter.next(), MessageChannel.class);
Flux.from((Publisher) publisher).doOnNext(message -> outputChannel.send((Message) message)).subscribe();
});
}
else {
outputBindingNames.stream().forEach(outputBindingName -> {
MessageChannel outputChannel = context.getBean(outputBindingName, MessageChannel.class);
Flux.from((Publisher) resultPublishers).doOnNext(message -> outputChannel.send((Message) message)).subscribe();
});
}
}
private void postProcessForStandAloneFunction(FunctionInvocationWrapper function, MessageChannel inputChannel) {
Type functionType = FunctionTypeUtils.getFunctionType(function, this.functionInspector);
if (FunctionTypeUtils.isReactive(FunctionTypeUtils.getInputType(functionType, 0))) {
MessageChannel outputChannel = context.getBean(Source.OUTPUT, MessageChannel.class);
SubscribableChannel subscribeChannel = (SubscribableChannel) inputChannel;
Publisher<?> publisher = this.enhancePublisher(MessageChannelReactiveUtils.toPublisher(subscribeChannel));
this.subscribeToInput(function, publisher, outputChannel::send);
//
private void bindOrComposeSimpleFunctions(String channelName, SubscribableChannel messageChannel,
BindableProxyFactory bindableProxyFactory, String functionDefinition) {
//TODO there is something about moving channel interceptors in AMCB (not sure if it is still required)
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);
FunctionInvocationWrapper function = functionCatalog.lookup(functionDefinition, properties.getContentType());
this.composeSimpleFunctionToExistingFlow(function, messageChannel, channelName, bindableProxyFactory);
}
else {
ServiceActivatingHandler handler = new ServiceActivatingHandler(new FunctionWrapper(function));
handler.setBeanFactory(context);
handler.afterPropertiesSet();
if (!FunctionTypeUtils.isConsumer(functionType)) {
handler.setOutputChannelName(Source.OUTPUT);
}
SubscribableChannel subscribeChannel = (SubscribableChannel) inputChannel;
subscribeChannel.subscribe(handler);
BindingProperties properties = this.serviceProperties.getBindings().get(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);
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);
handler.setOutputChannelName("output.extended");
messageChannel.subscribe(handler);
}
private void bindSimpleFunctions(FunctionInvocationWrapper function, SubscribableChannel inputChannel, BindableProxyFactory bindableProxyFactory) {
Type functionType = FunctionTypeUtils.getFunctionType(function, this.functionInspector);
String outputChannelName = bindableProxyFactory instanceof BindableFunctionProxyFactory
? ((BindableFunctionProxyFactory) bindableProxyFactory).getOutputName(0)
: Source.OUTPUT;
if (FunctionTypeUtils.isReactive(FunctionTypeUtils.getInputType(functionType, 0))) {
MessageChannel outputChannel = context.getBean(outputChannelName, MessageChannel.class);
SubscribableChannel subscribeChannel = (SubscribableChannel) inputChannel;
Publisher<?> publisher = this.enhancePublisher(MessageChannelReactiveUtils.toPublisher(subscribeChannel),
((DirectWithAttributesChannel) inputChannel).getBeanName());
this.subscribeToInput(function, publisher, message -> outputChannel.send((Message<?>) message));
}
else {
ServiceActivatingHandler handler = createFunctionHandler(function);
if (!FunctionTypeUtils.isConsumer(functionType)) {
handler.setOutputChannelName(outputChannelName);
}
inputChannel.subscribe(handler);
}
}
private ServiceActivatingHandler createFunctionHandler(FunctionInvocationWrapper function) {
ServiceActivatingHandler handler = new ServiceActivatingHandler(new FunctionWrapper(function));
handler.setBeanFactory(context);
handler.afterPropertiesSet();
return handler;
}
/*
* Enhance publisher to add error handling, retries etc.
*/
@SuppressWarnings({ "unchecked", "rawtypes" })
private Publisher enhancePublisher(Publisher publisher) {
private Publisher enhancePublisher(Publisher publisher, String bindingName) {
Flux flux = Flux.from(publisher)
.concatMap(message -> {
ConsumerProperties consumerProperties = this.serviceProperties.getBindings().get(Sink.INPUT).getConsumer();
ConsumerProperties consumerProperties = this.serviceProperties.getBindings().get(bindingName).getConsumer();
return Flux.just(message)
.doOnError(e -> {
e.printStackTrace();
@@ -349,27 +425,60 @@ public class FunctionConfiguration {
@SuppressWarnings({ "unchecked", "rawtypes" })
private <I, O> void subscribeToInput(Function function,
Publisher<?> publisher, Consumer<Message<O>> outputProcessor) {
Function<Flux<Message<I>>, Flux<Message<O>>> functionInvoker = function;
private void subscribeToInput(Function function, Publisher publisher, Consumer<Message> outputProcessor) {
Function<Flux<Message>, Flux<Message>> functionInvoker = function;
Flux<?> inputPublisher = Flux.from(publisher);
subscribeToOutput(outputProcessor,
functionInvoker.apply((Flux<Message<I>>) inputPublisher)).subscribe();
subscribeToOutput(outputProcessor, functionInvoker.apply((Flux<Message>) inputPublisher)).subscribe();
}
private <O> Mono<Void> subscribeToOutput(Consumer<Message<O>> outputProcessor,
Publisher<Message<O>> outputPublisher) {
Flux<Message<O>> output = outputProcessor == null ? Flux.from(outputPublisher)
: Flux.from(outputPublisher).doOnNext(outputProcessor);
@SuppressWarnings("rawtypes")
private Mono<Void> subscribeToOutput(Consumer<Message> outputProcessor, Flux<Message> resultPublisher) {
Flux<Message> output = outputProcessor == null
? resultPublisher
: resultPublisher.doOnNext(outputProcessor);
return output.then();
}
private void assertSupportedSignatures(Type functionType) {
Assert.isTrue(!FunctionTypeUtils.isConsumer(functionType),
"Function '" + functionProperties.getDefinition() + "' is a Consumer which is not supported "
+ "for multi-in/out reactive streams. Only Functions are supported");
Assert.isTrue(!FunctionTypeUtils.isSupplier(functionType),
"Function '" + functionProperties.getDefinition() + "' is a Supplier which is not supported "
+ "for multi-in/out reactive streams. Only Functions are supported");
Assert.isTrue(!FunctionTypeUtils.isInputArray(functionType) && !FunctionTypeUtils.isOutputArray(functionType),
"Function '" + functionProperties.getDefinition() + "' has the following signature: ["
+ functionType + "]. Your input and/or outout lacks arity and therefore we "
+ "can not determine how many input/output destinations are required in the context of "
+ "function input/output binding.");
int inputCount = FunctionTypeUtils.getInputCount(functionType);
for (int i = 0; i < inputCount; i++) {
Assert.isTrue(FunctionTypeUtils.isReactive(FunctionTypeUtils.getInputType(functionType, i)),
"Function '" + functionProperties.getDefinition() + "' has the following signature: ["
+ functionType + "]. Non-reactive functions with multiple "
+ "inputs/outputs are not supported in the context of Spring Cloud Stream.");
}
int outputCount = FunctionTypeUtils.getOutputCount(functionType);
for (int i = 0; i < outputCount; i++) {
Assert.isTrue(FunctionTypeUtils.isReactive(FunctionTypeUtils.getInputType(functionType, i)),
"Function '" + functionProperties.getDefinition() + "' has the following signature: ["
+ functionType + "]. Non-reactive functions with multiple "
+ "inputs/outputs are not supported in the context of Spring Cloud Stream.");
}
}
private boolean isMultipleInputOutput(BindableProxyFactory bindableProxyFactory) {
return bindableProxyFactory instanceof BindableFunctionProxyFactory
&& ((BindableFunctionProxyFactory) bindableProxyFactory).isMultiple();
}
}
/**
*
* Ensure that SI does not attempt any conversion and sends a raw Message.
*
* It's signatures ensures that within the context of s-c-stream Spring Integration does
* not attempt any conversion and sends a raw Message.
*/
@SuppressWarnings("rawtypes")
private static class FunctionWrapper implements Function<Message<byte[]>, Object> {
@@ -389,11 +498,11 @@ public class FunctionConfiguration {
}
}
/*
* This class will effectively create a different representation of Bindable interfaces (e.g., Source, Processor...).
* It's main goal is to determine the count of inputs and outputs based on the provided function.
/**
* Creates and registers instances of BindableFunctionProxyFactory for each user defined function
* thus triggering destination bindings between function arguments and destinations.
*/
private static class FunctionBindingHolder implements InitializingBean, ApplicationContextAware, EnvironmentAware {
private static class FunctionBindingRegistrar implements InitializingBean, ApplicationContextAware, EnvironmentAware {
private final BinderTypeRegistry binderTypeRegistry;
@@ -409,7 +518,7 @@ public class FunctionConfiguration {
private int outputCount;
FunctionBindingHolder(BinderTypeRegistry binderTypeRegistry, FunctionCatalog functionCatalog, StreamFunctionProperties streamFunctionProperties) {
FunctionBindingRegistrar(BinderTypeRegistry binderTypeRegistry, FunctionCatalog functionCatalog, StreamFunctionProperties streamFunctionProperties) {
this.binderTypeRegistry = binderTypeRegistry;
this.functionCatalog = functionCatalog;
this.streamFunctionProperties = streamFunctionProperties;
@@ -426,37 +535,35 @@ public class FunctionConfiguration {
&& ObjectUtils.isEmpty(applicationContext.getBeanNamesForAnnotation(EnableBinding.class))
&& this.determineFunctionName(functionCatalog, environment)) {
BeanDefinitionRegistry registry = (BeanDefinitionRegistry) applicationContext.getBeanFactory();
RootBeanDefinition rootBeanDefinition = new RootBeanDefinition(BindableFunctionProxyFactory.class);
FunctionInvocationWrapper function = functionCatalog.lookup(streamFunctionProperties.getDefinition());
if (function != null) {
if (function.isSupplier()) {
this.inputCount = 0;
this.outputCount = 1;
String[] functionDefinitions = streamFunctionProperties.getDefinition().split(";");
boolean nameBasedOnFunctionName = functionDefinitions.length > 1;
for (String functionDefinition : functionDefinitions) {
RootBeanDefinition functionBindableProxyDefinition = new RootBeanDefinition(BindableFunctionProxyFactory.class);
FunctionInvocationWrapper function = functionCatalog.lookup(functionDefinition);
if (function != null) {
Type functionType = function.getFunctionType();
if (function.isSupplier()) {
this.inputCount = 0;
this.outputCount = FunctionTypeUtils.getOutputCount(functionType);
}
else if (function.isConsumer()) {
this.inputCount = FunctionTypeUtils.getInputCount(functionType);
this.outputCount = 0;
}
else {
this.inputCount = FunctionTypeUtils.getInputCount(functionType);
this.outputCount = FunctionTypeUtils.getOutputCount(functionType);
}
functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(functionDefinition);
functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(this.inputCount);
functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(this.outputCount);
functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(nameBasedOnFunctionName);
registry.registerBeanDefinition(functionDefinition + "_binding", functionBindableProxyDefinition);
}
else if (function.isConsumer()) {
this.inputCount = 1;
this.outputCount = 0;
}
else {
this.inputCount = 1;
this.outputCount = 1;
}
rootBeanDefinition.getConstructorArgumentValues().addGenericArgumentValue(this.inputCount);
rootBeanDefinition.getConstructorArgumentValues().addGenericArgumentValue(this.outputCount);
registry.registerBeanDefinition(streamFunctionProperties.getDefinition() + "_binding",
rootBeanDefinition);
}
}
}
int getInputCount() {
return this.inputCount;
}
int getOutputCount() {
return this.outputCount;
}
private boolean determineFunctionName(FunctionCatalog catalog, Environment environment) {
String definition = streamFunctionProperties.getDefinition();
if (!StringUtils.hasText(definition)) {

15
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/test/AbstractDestination.java
View File

@@ -16,6 +16,9 @@
package org.springframework.cloud.stream.binder.test;
import java.util.ArrayList;
import java.util.List;
import org.springframework.messaging.SubscribableChannel;
/**
@@ -24,18 +27,18 @@ import org.springframework.messaging.SubscribableChannel;
*/
abstract class AbstractDestination {
private SubscribableChannel channel;
private final List<SubscribableChannel> channels = new ArrayList<>();
SubscribableChannel getChannel() {
return this.channel;
SubscribableChannel getChannel(int index) {
return this.channels.get(index);
}
void setChannel(SubscribableChannel channel) {
this.channel = channel;
this.afterChannelIsSet();
this.channels.add(channel);
this.afterChannelIsSet(this.channels.size() - 1);
}
void afterChannelIsSet() {
void afterChannelIsSet(int channelIndex) {
// noop
}

6
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/test/InputDestination.java
View File

@@ -34,7 +34,11 @@ public class InputDestination extends AbstractDestination {
* @param message message to send
*/
public void send(Message<?> message) {
this.getChannel().send(message);
this.getChannel(0).send(message);
}
public void send(Message<?> message, int inputIndex) {
this.getChannel(inputIndex).send(message);
}
}

21
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/test/OutputDestination.java
View File

@@ -16,6 +16,8 @@
package org.springframework.cloud.stream.binder.test;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedTransferQueue;
import java.util.concurrent.TimeUnit;
@@ -32,7 +34,7 @@ import org.springframework.messaging.Message;
*/
public class OutputDestination extends AbstractDestination {
private BlockingQueue<Message<?>> messages;
private final List<BlockingQueue<Message<?>>> messageQueues = new ArrayList<>();
/**
* Allows to access {@link Message}s received by this {@link OutputDestination}.
@@ -40,9 +42,9 @@ public class OutputDestination extends AbstractDestination {
* @return received message
*/
@SuppressWarnings("unchecked")
public Message<byte[]> receive(long timeout) {
public Message<byte[]> receive(long timeout, int channelIndex) {
try {
return (Message<byte[]>) this.messages.poll(timeout, TimeUnit.MILLISECONDS);
return (Message<byte[]>) this.messageQueues.get(channelIndex).poll(timeout, TimeUnit.MILLISECONDS);
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
@@ -55,13 +57,18 @@ public class OutputDestination extends AbstractDestination {
* @return received message
*/
public Message<byte[]> receive() {
return this.receive(0);
return this.receive(0, 0);
}
public Message<byte[]> receive(long timeout) {
return this.receive(timeout, 0);
}
@Override
void afterChannelIsSet() {
this.messages = new LinkedTransferQueue<>();
this.getChannel().subscribe(message -> this.messages.offer(message));
void afterChannelIsSet(int channelIndex) {
BlockingQueue<Message<?>> messageQueue = new LinkedTransferQueue<>();
this.messageQueues.add(messageQueue);
this.getChannel(channelIndex).subscribe(message -> this.messageQueues.get(channelIndex).offer(message));
}
}

292
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/MultipleInputOutputFunctionTests.java Normal file
View File

@@ -0,0 +1,292 @@
/*
* Copyright 2019-2019 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
*
* https://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 java.util.function.Function;
import org.junit.Test;
import reactor.core.publisher.Flux;
import reactor.core.publisher.UnicastProcessor;
import reactor.util.function.Tuple2;
import reactor.util.function.Tuples;
import org.springframework.beans.factory.BeanCreationException;
import org.springframework.boot.WebApplicationType;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.builder.SpringApplicationBuilder;
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.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import static org.assertj.core.api.Assertions.assertThat;
/**
*
* @author Oleg Zhurakousky
*
*/
public class MultipleInputOutputFunctionTests {
@Test(expected = BeanCreationException.class)
public void testFailureWithNonReactiveFunction() {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(
ReactiveFunctionConfiguration.class))
.web(WebApplicationType.NONE)
.run("--spring.jmx.enabled=false",
"--spring.cloud.function.definition=multipleInputNonReactive")) {
context.getBean(InputDestination.class);
}
}
@Test(expected = BeanCreationException.class)
public void testFailureWithReactiveArrayOutput() {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(
ReactiveFunctionConfiguration.class))
.web(WebApplicationType.NONE)
.run("--spring.jmx.enabled=false",
"--spring.cloud.function.definition=multiReactiveInputReactiveArrayOutput")) {
context.getBean(InputDestination.class);
}
}
@Test(expected = BeanCreationException.class)
public void testFailureWithReactiveArrayOutputNonGeneric() {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(
ReactiveFunctionConfiguration.class))
.web(WebApplicationType.NONE)
.run("--spring.jmx.enabled=false",
"--spring.cloud.function.definition=multiReactiveInputReactiveArrayOutputNoGeneric")) {
context.getBean(InputDestination.class);
}
}
@Test(expected = BeanCreationException.class)
public void testFailureWithReactiveArrayInput() {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(
ReactiveFunctionConfiguration.class))
.web(WebApplicationType.NONE)
.run("--spring.jmx.enabled=false",
"--spring.cloud.function.definition=genericReactiveArrayInput")) {
context.getBean(InputDestination.class);
}
}
@Test(expected = BeanCreationException.class)
public void testFailureWithReactiveArrayInputNonGeneric() {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(
ReactiveFunctionConfiguration.class))
.web(WebApplicationType.NONE)
.run("--spring.jmx.enabled=false",
"--spring.cloud.function.definition=nonGenericReactiveArrayInput")) {
context.getBean(InputDestination.class);
}
}
@Test(expected = BeanCreationException.class)
public void testFailureWithConsumer() {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(
ReactiveFunctionConfiguration.class))
.web(WebApplicationType.NONE)
.run("--spring.jmx.enabled=false",
"--spring.cloud.function.definition=multiInputConsumer")) {
context.getBean(InputDestination.class);
}
}
@Test
public void testMultiInputSingleOutput() {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(
ReactiveFunctionConfiguration.class))
.web(WebApplicationType.NONE)
.run("--spring.jmx.enabled=false",
"--spring.cloud.function.definition=multiInputSingleOutput")) {
context.getBean(InputDestination.class);
InputDestination inputDestination = context.getBean(InputDestination.class);
OutputDestination outputDestination = context.getBean(OutputDestination.class);
Message<byte[]> stringInputMessage = MessageBuilder.withPayload("one".getBytes()).build();
Message<byte[]> integerInputMessage = MessageBuilder.withPayload("1".getBytes()).build();
inputDestination.send(stringInputMessage, 0);
inputDestination.send(integerInputMessage, 1);
Message<byte[]> outputMessage = outputDestination.receive();
assertThat(outputMessage.getPayload()).isEqualTo("one".getBytes());
outputMessage = outputDestination.receive();
assertThat(outputMessage.getPayload()).isEqualTo("1".getBytes());
}
}
@Test
public void testSingleInputMultiOutput() {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(
ReactiveFunctionConfiguration.class))
.web(WebApplicationType.NONE)
.run("--spring.jmx.enabled=false",
"--spring.cloud.function.definition=singleInputMultipleOutputs")) {
context.getBean(InputDestination.class);
InputDestination inputDestination = context.getBean(InputDestination.class);
OutputDestination outputDestination = context.getBean(OutputDestination.class);
for (int i = 0; i < 10; i++) {
inputDestination.send(MessageBuilder.withPayload(String.valueOf(i).getBytes()).build());
}
int counter = 0;
for (int i = 0; i < 5; i++) {
Message<byte[]> even = outputDestination.receive(0, 0);
assertThat(even.getPayload()).isEqualTo(("EVEN: " + String.valueOf(counter++)).getBytes());
Message<byte[]> odd = outputDestination.receive(0, 1);
assertThat(odd.getPayload()).isEqualTo(("ODD: " + String.valueOf(counter++)).getBytes());
}
}
}
@Test
public void testMultipleFunctions() {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(
ReactiveFunctionConfiguration.class))
.web(WebApplicationType.NONE)
.run("--spring.jmx.enabled=false",
"--spring.cloud.function.definition=uppercase;reverse")) {
context.getBean(InputDestination.class);
InputDestination inputDestination = context.getBean(InputDestination.class);
OutputDestination outputDestination = context.getBean(OutputDestination.class);
Message<byte[]> inputMessage = MessageBuilder.withPayload("Hello".getBytes()).build();
inputDestination.send(inputMessage, 0);
inputDestination.send(inputMessage, 1);
Message<byte[]> outputMessage = outputDestination.receive(0, 0);
assertThat(outputMessage.getPayload()).isEqualTo("HELLO".getBytes());
outputMessage = outputDestination.receive(0, 1);
assertThat(outputMessage.getPayload()).isEqualTo("olleH".getBytes());
}
}
@Test
public void testMultipleFunctionsWithComposition() {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(
ReactiveFunctionConfiguration.class))
.web(WebApplicationType.NONE)
.run("--spring.jmx.enabled=false",
"--spring.cloud.function.definition=uppercase|reverse;reverse|uppercase")) {
context.getBean(InputDestination.class);
InputDestination inputDestination = context.getBean(InputDestination.class);
OutputDestination outputDestination = context.getBean(OutputDestination.class);
Message<byte[]> inputMessage = MessageBuilder.withPayload("Hello".getBytes()).build();
inputDestination.send(inputMessage, 0);
inputDestination.send(inputMessage, 1);
Message<byte[]> outputMessage = outputDestination.receive(0, 0);
assertThat(outputMessage.getPayload()).isEqualTo("OLLEH".getBytes());
outputMessage = outputDestination.receive(0, 1);
assertThat(outputMessage.getPayload()).isEqualTo("OLLEH".getBytes());
}
}
@EnableAutoConfiguration
public static class ReactiveFunctionConfiguration {
@Bean
public Function<String, String> uppercase() {
return value -> value.toUpperCase();
}
@Bean
public Function<String, String> reverse() {
return value -> new StringBuilder(value).reverse().toString();
}
@Bean
public Function<Tuple2<String, String>, String> multipleInputNonReactive() { // not supported
return tuple -> null;
}
@Bean
public Function<Tuple2<Flux<String>, Flux<Integer>>, Flux<?>[]> multiReactiveInputReactiveArrayOutput() { // not supported
return tuple -> null;
}
@Bean
public Consumer<Tuple2<Flux<String>, Flux<Integer>>> multiInputConsumer() { // not supported
return tuple -> System.out.println();
}
@SuppressWarnings("rawtypes")
@Bean
public Function<Tuple2<Flux<String>, Flux<Integer>>, Flux[]> multiReactiveInputReactiveArrayOutputNoGeneric() { // not supported
return tuple -> null;
}
@Bean
public Function<Flux<?>[], Tuple2<Flux<String>, Flux<Integer>>> genericReactiveArrayInput() { // not supported
return tuple -> null;
}
@SuppressWarnings("rawtypes")
@Bean
public Function<Flux[], Tuple2<Flux<String>, Flux<Integer>>> nonGenericReactiveArrayInput() { // not supported
return tuple -> null;
}
@Bean
public Function<Tuple2<Flux<String>, Flux<Integer>>, Flux<String>> multiInputSingleOutput() {
return tuple -> {
Flux<String> stringStream = tuple.getT1();
Flux<String> intStream = tuple.getT2().map(i -> String.valueOf(i));
return Flux.merge(stringStream, intStream);
};
}
@Bean
@SuppressWarnings({ "unchecked", "rawtypes" })
public static Function<Flux<Integer>, Tuple2<Flux<String>, Flux<String>>> singleInputMultipleOutputs() {
return flux -> {
Flux<Integer> connectedFlux = flux.publish().autoConnect(2);
UnicastProcessor even = UnicastProcessor.create();
UnicastProcessor odd = UnicastProcessor.create();
Flux<Integer> evenFlux = connectedFlux.filter(number -> number % 2 == 0).doOnNext(number -> even.onNext("EVEN: " + number));
Flux<Integer> oddFlux = connectedFlux.filter(number -> number % 2 != 0).doOnNext(number -> odd.onNext("ODD: " + number));
return Tuples.of(Flux.from(even).doOnSubscribe(x -> evenFlux.subscribe()), Flux.from(odd).doOnSubscribe(x -> oddFlux.subscribe()));
};
}
}
}

1
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/SourceToFunctionsSupportTests.java
View File

@@ -234,7 +234,6 @@ public class SourceToFunctionsSupportTests {
assertThat(new String(target.receive(2000).getPayload())).isEqualTo("6");
assertThat(context.getBean("supplierInitializer")).isNotEqualTo(null);
assertThat(context.getBean("functionInitializer")).isEqualTo(null);
}
}