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

GH-1917 add missing support for composing reactive function

Browse Source 
Added missing support for composing reactive function to an existing legacy source. Imperative functions already had such support
Added tests as well as minor refactoring/cleanup in FunctionConfiguration

Resolves #1917
This commit is contained in:
Oleg Zhurakousky
2020-02-26 09:51:11 +01:00
parent b5e9444d0c
commit f25d37683f
2 changed files with 50 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -21,6 +21,7 @@ import java.lang.reflect.Method;
import java.lang.reflect.Type;
import java.time.Instant;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
@@ -391,69 +392,54 @@ public class FunctionConfiguration {
Type functionType = function.getFunctionType();
this.assertSupportedSignatures(bindableProxyFactory, functionType);
if (this.functionProperties.isComposeFrom()) {
AbstractSubscribableChannel outputChannel = this.applicationContext.getBean(outputBindingNames.iterator().next(), AbstractSubscribableChannel.class);
logger.info("Composing at the head of output destination: " + outputChannel.getBeanName());
String outputChannelName = ((AbstractMessageChannel) outputChannel).getBeanName();
DirectWithAttributesChannel newOutputChannel = new DirectWithAttributesChannel();
newOutputChannel.setAttribute("type", "output");
newOutputChannel.setComponentName("output.extended");
this.applicationContext.registerBean("output.extended", MessageChannel.class, () -> newOutputChannel);
bindableProxyFactory.replaceOutputChannel(outputChannelName, "output.extended", newOutputChannel);
inputBindingNames = Collections.singleton("output");
}
if (isReactiveOrMultipleInputOutput(bindableProxyFactory, functionType)) {
Publisher[] inputPublishers = inputBindingNames.stream().map(inputBindingName -> {
SubscribableChannel inputChannel = this.applicationContext.getBean(inputBindingName, SubscribableChannel.class);
return MessageChannelReactiveUtils.toPublisher(inputChannel);
}).toArray(Publisher[]::new);
ProducerProperties producerProperties = this.serviceProperties.getBindings().get(outputBindingNames.iterator().next()).getProducer();
BindingProperties bindingProperties = this.serviceProperties.getBindings().get(outputBindingNames.iterator().next());
ProducerProperties producerProperties = bindingProperties == null ? null : bindingProperties.getProducer();
PartitionAwareFunction functionToInvoke = new PartitionAwareFunction(function, this.applicationContext, producerProperties);
Object resultPublishers = functionToInvoke.apply(inputPublishers.length == 1 ? inputPublishers[0] : Tuples.fromArray(inputPublishers));
if (resultPublishers instanceof Iterable) {
Iterator<String> outputBindingIter = outputBindingNames.iterator();
((Iterable) resultPublishers).forEach(publisher -> {
MessageChannel outputChannel = this.applicationContext.getBean(outputBindingIter.next(), MessageChannel.class);
Flux.from((Publisher) publisher)
.onErrorContinue((ex, pay) -> {
logger.error("Failed to process the following content which will be dropped: " + pay, (Throwable) ex);
})
.doOnNext(message -> outputChannel.send((Message) message)).subscribe();
});
}
else {
outputBindingNames.stream().forEach(outputBindingName -> {
MessageChannel outputChannel = this.applicationContext.getBean(outputBindingName, MessageChannel.class);
Flux.from((Publisher) resultPublishers)
.onErrorContinue((ex, pay) -> {
logger.error("Failed to process the following content which will be dropped: " + pay, (Throwable) ex);
})
.doOnNext(message -> {
outputChannel.send((Message) message);
}).subscribe();
});
if (!(resultPublishers instanceof Iterable)) {
resultPublishers = Collections.singletonList(resultPublishers);
}
Iterator<String> outputBindingIter = outputBindingNames.iterator();
((Iterable) resultPublishers).forEach(publisher -> {
MessageChannel outputChannel = this.applicationContext.getBean(outputBindingIter.next(), MessageChannel.class);
Flux.from((Publisher) publisher)
.onErrorContinue((ex, pay) -> {
logger.error("Failed to process the following content which will be dropped: " + pay, (Throwable) ex);
})
.doOnNext(message -> outputChannel.send((Message) message)).subscribe();
});
}
else {
String outputDestinationName = this.determineOutputDestinationName(0, bindableProxyFactory, functionType);
this.adjustFunctionForNativeEncodingIfNecessary(outputDestinationName, function, 0);
if (this.functionProperties.isComposeFrom()) {
AbstractSubscribableChannel outputChannel = this.applicationContext.getBean(outputDestinationName, AbstractSubscribableChannel.class);
logger.info("Composing at the head of output destination: " + outputChannel.getBeanName());
String outputChannelName = ((AbstractMessageChannel) outputChannel).getBeanName();
ServiceActivatingHandler handler = createFunctionHandler(function, null, outputChannelName);
DirectWithAttributesChannel newOutputChannel = new DirectWithAttributesChannel();
newOutputChannel.setAttribute("type", "output");
newOutputChannel.setComponentName("output.extended");
this.applicationContext.registerBean("output.extended", MessageChannel.class, () -> newOutputChannel);
bindableProxyFactory.replaceOutputChannel(outputChannelName, "output.extended", newOutputChannel);
handler.setOutputChannelName("output.extended");
outputChannel.subscribe(handler);
}
else {
String inputDestinationName = inputBindingNames.iterator().next();
Object inputDestination = this.applicationContext.getBean(inputDestinationName);
if (inputDestination != null && inputDestination instanceof SubscribableChannel) {
ServiceActivatingHandler handler = createFunctionHandler(function, inputDestinationName, outputDestinationName);
if (!FunctionTypeUtils.isConsumer(function.getFunctionType())) {
handler.setOutputChannelName(outputDestinationName);
}
((SubscribableChannel) inputDestination).subscribe(handler);
String inputDestinationName = inputBindingNames.iterator().next();
Object inputDestination = this.applicationContext.getBean(inputDestinationName);
if (inputDestination != null && inputDestination instanceof SubscribableChannel) {
ServiceActivatingHandler handler = createFunctionHandler(function, inputDestinationName, outputDestinationName);
if (!FunctionTypeUtils.isConsumer(function.getFunctionType())) {
handler.setOutputChannelName(outputDestinationName);
}
((SubscribableChannel) inputDestination).subscribe(handler);
}
}
}
@@ -516,9 +502,10 @@ public class FunctionConfiguration {
}
private String determineOutputDestinationName(int index, BindableProxyFactory bindableProxyFactory, Type functionType) {
List<String> outputNames = new ArrayList<>(bindableProxyFactory.getOutputs());
String outputDestinationName = bindableProxyFactory instanceof BindableFunctionProxyFactory
? ((BindableFunctionProxyFactory) bindableProxyFactory).getOutputName(index)
: (FunctionTypeUtils.isConsumer(functionType) ? null : "output");
: (FunctionTypeUtils.isConsumer(functionType) ? null : outputNames.get(index));
return outputDestinationName;
}
@@ -593,10 +580,6 @@ public class FunctionConfiguration {
isRoutingFunction = ((FunctionInvocationWrapper) function).getTarget() instanceof RoutingFunction;
this.applicationContext = applicationContext;
this.function = new PartitionAwareFunction((FunctionInvocationWrapper) function, this.applicationContext, producerProperties);
Type type = ((FunctionInvocationWrapper) function).getFunctionType();
if (FunctionTypeUtils.isReactive(FunctionTypeUtils.getOutputType(type, 0))) {
//throw new IllegalStateException("Functions with mixed semantics (imperative input vs. reactive output) ar not supported at the moment");
}
this.consumerProperties = consumerProperties;
this.producerProperties = producerProperties;
this.headersField = ReflectionUtils.findField(MessageHeaders.class, "headers");

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

@@ -84,6 +84,20 @@ public class SourceToFunctionsSupportTests {
}
}
@Test
public void testFunctionsAreAppliedToExistingMessageSourceReactive() {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(
FunctionsConfiguration.class, ExistingMessageSourceConfiguration.class)).web(WebApplicationType.NONE).run(
"--spring.cloud.stream.function.definition=|toUpperCaseReactive|concatWithSelf",
"--spring.jmx.enabled=false")) {
OutputDestination target = context.getBean(OutputDestination.class);
assertThat(target.receive(1000).getPayload())
.isEqualTo("HELLO FUNCTION:HELLO FUNCTION".getBytes(StandardCharsets.UTF_8));
}
}
@Test
public void testImperativeSupplier() {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(