From c614e15f1042632189fe03495e2920d4bf106a08 Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Tue, 21 Jan 2020 16:03:50 +0100 Subject: [PATCH] Additional consolidation and simplification of logic around bootstrapping multiple argument and single arguments functions --- .../function/FunctionConfiguration.java | 101 ++++++++++++++---- 1 file changed, 81 insertions(+), 20 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 d1d84d437..3031c854f 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 @@ -19,6 +19,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; import java.time.Instant; import java.util.ArrayList; import java.util.Iterator; @@ -26,6 +27,7 @@ import java.util.List; import java.util.Map; import java.util.Set; import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Consumer; import java.util.function.Function; import java.util.function.Supplier; @@ -92,6 +94,7 @@ import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.SubscribableChannel; +import org.springframework.messaging.support.ErrorMessage; import org.springframework.scheduling.TaskScheduler; import org.springframework.util.Assert; import org.springframework.util.ClassUtils; @@ -335,23 +338,35 @@ public class FunctionConfiguration { this.assertSupportedSignatures(bindableProxyFactory, functionType); if (isReactiveOrMultipleInputOutput(bindableProxyFactory, functionType)) { - Publisher[] inputPublishers = inputBindingNames.stream().map(inputBindingName -> { + + if (!this.isMultipleInputOutput(bindableProxyFactory)) { + String inputBindingName = inputBindingNames.iterator().next(); + String outputBindingName = outputBindingNames.iterator().next(); SubscribableChannel inputChannel = this.applicationContext.getBean(inputBindingName, SubscribableChannel.class); - return MessageChannelReactiveUtils.toPublisher(inputChannel); - }).toArray(Publisher[]::new); - Object resultPublishers = function.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).doOnNext(message -> outputChannel.send((Message) message)).subscribe(); - }); + MessageChannel outputChannel = this.applicationContext.getBean(outputBindingName, MessageChannel.class); + this.subscribeToInput(function, inputBindingName, MessageChannelReactiveUtils.toPublisher(inputChannel), + message -> outputChannel.send(message)); } else { - outputBindingNames.stream().forEach(outputBindingName -> { - MessageChannel outputChannel = this.applicationContext.getBean(outputBindingName, MessageChannel.class); - Flux.from((Publisher) resultPublishers).doOnNext(message -> outputChannel.send((Message) message)).subscribe(); - }); + Publisher[] inputPublishers = inputBindingNames.stream().map(inputBindingName -> { + SubscribableChannel inputChannel = this.applicationContext.getBean(inputBindingName, SubscribableChannel.class); + return MessageChannelReactiveUtils.toPublisher(inputChannel); + }).toArray(Publisher[]::new); + + Object resultPublishers = function.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).doOnNext(message -> outputChannel.send((Message) message)).subscribe(); + }); + } + else { + outputBindingNames.stream().forEach(outputBindingName -> { + MessageChannel outputChannel = this.applicationContext.getBean(outputBindingName, MessageChannel.class); + Flux.from((Publisher) resultPublishers).doOnNext(message -> outputChannel.send((Message) message)).subscribe(); + }); + } } } else { @@ -374,17 +389,63 @@ public class FunctionConfiguration { } else { String inputDestinationName = inputBindingNames.iterator().next(); - //this.adjustFunctionForNativeEncodingIfNecessary(); - ServiceActivatingHandler handler = createFunctionHandler(function, inputDestinationName, outputDestinationName); - if (!FunctionTypeUtils.isConsumer(function.getFunctionType())) { - handler.setOutputChannelName(outputDestinationName); + 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); } - SubscribableChannel inputChannel = this.applicationContext.getBean(inputDestinationName, SubscribableChannel.class); - inputChannel.subscribe(handler); } } } + @SuppressWarnings({ "unchecked", "rawtypes" }) + private void subscribeToInput(Function function, String bindingName, Publisher publisher, Consumer outputProcessor) { + + Flux inputPublisher = Flux.from(publisher); + + AtomicReference> originalMessageRef = new AtomicReference<>(); + AtomicReference consumerPropertiesRef = new AtomicReference<>(); + AtomicReference bindingErrorChannelRef = + new AtomicReference<>(this.applicationContext.getBean("errorChannel", MessageChannel.class)); + + Flux result = inputPublisher + .switchOnFirst((x, message) -> { + consumerPropertiesRef.set(this.serviceProperties.getBindings().get(bindingName).getConsumer()); + String destination = serviceProperties.getBindings().get(bindingName).getDestination(); + String group = serviceProperties.getBindings().get(bindingName).getGroup(); + String bindingErrorChannelName = destination + "." + group + ".errors"; + if (this.applicationContext.containsBean(bindingErrorChannelName)) { + bindingErrorChannelRef.set(this.applicationContext.getBean(bindingErrorChannelName, MessageChannel.class)); + } + return message; + }) + .concatMap(message -> { + return Flux.just(message).doOnNext(originalMessageRef::set) + .transform((Function, Flux>) function) + .retryBackoff(consumerPropertiesRef.get().getMaxAttempts(), + Duration.ofMillis(consumerPropertiesRef.get().getBackOffInitialInterval()), + Duration.ofMillis(consumerPropertiesRef.get().getBackOffMaxInterval())) + .onErrorResume(e -> { + bindingErrorChannelRef.get() + .send(new ErrorMessage((Throwable) e, originalMessageRef.get().getHeaders(), originalMessageRef.get())); + return Mono.empty(); + }); + }); + + subscribeToOutput(outputProcessor, result).subscribe(); + } + + @SuppressWarnings("rawtypes") + private Mono subscribeToOutput(Consumer outputProcessor, Flux resultPublisher) { + Flux output = outputProcessor == null + ? resultPublisher + : resultPublisher.doOnNext(outputProcessor); + return output.then(); + } + private void adjustFunctionForNativeEncodingIfNecessary(String outputDestinationName, FunctionInvocationWrapper function, int index) { if (function.isConsumer()) { return;