From f25d37683fea67d0d2a01f23ac2f30668d592399 Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Wed, 26 Feb 2020 09:51:11 +0100 Subject: [PATCH] GH-1917 add missing support for composing reactive function 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 --- .../function/FunctionConfiguration.java | 89 ++++++++----------- .../SourceToFunctionsSupportTests.java | 14 +++ 2 files changed, 50 insertions(+), 53 deletions(-) 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 06369699f..c907eb053 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 @@ -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 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 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 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"); diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/SourceToFunctionsSupportTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/SourceToFunctionsSupportTests.java index c319d6d4a..b77c752d8 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/SourceToFunctionsSupportTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/SourceToFunctionsSupportTests.java @@ -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(