From 387100f107bd088bebde1eea795c5b117040157a Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Fri, 8 May 2020 14:47:25 +0200 Subject: [PATCH] Improve dynamic destination resolution Added StreamBridge.sendLight method which effectively does everything that 'send' does but maintains full control as to what is being cached and how. --- .../cloud/stream/binding/BindingService.java | 11 +++-- .../cloud/stream/function/StreamBridge.java | 42 ++++++++++++++++++- 2 files changed, 49 insertions(+), 4 deletions(-) diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java index 124662d0f..221b47a13 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java @@ -249,9 +249,8 @@ public class BindingService { } }); } - @SuppressWarnings({ "unchecked", "rawtypes" }) - public Binding bindProducer(T output, String outputName) { + public Binding bindProducer(T output, String outputName, boolean cache) { String bindingTarget = this.bindingServiceProperties .getBindingDestination(outputName); Binder binder = (Binder) getBinder( @@ -270,10 +269,16 @@ public class BindingService { validate(producerProperties); Binding binding = doBindProducer(output, bindingTarget, binder, producerProperties); - this.producerBindings.put(outputName, binding); + if (cache) { + this.producerBindings.put(outputName, binding); + } return binding; } + public Binding bindProducer(T output, String outputName) { + return this.bindProducer(output, outputName, true); + } + @SuppressWarnings("rawtypes") public Object getExtendedProducerProperties(Object output, String outputName) { Binder binder = getBinder(outputName, output.getClass()); diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java index 36f600ea4..40a7e8cc1 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java @@ -20,25 +20,38 @@ import java.util.HashMap; import java.util.Map; import java.util.Map.Entry; import java.util.function.Function; +import java.util.stream.Stream; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; - +import org.springframework.aop.framework.Advised; +import org.springframework.beans.BeanUtils; import org.springframework.beans.factory.SmartInitializingSingleton; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.cache.annotation.Cacheable; +import org.springframework.cache.annotation.EnableCaching; import org.springframework.cloud.function.context.FunctionCatalog; import org.springframework.cloud.function.context.FunctionRegistration; import org.springframework.cloud.function.context.FunctionRegistry; import org.springframework.cloud.function.context.FunctionType; import org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry.FunctionInvocationWrapper; +import org.springframework.cloud.stream.binder.Binder; +import org.springframework.cloud.stream.binder.Binding; +import org.springframework.cloud.stream.binder.ExtendedProducerProperties; +import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder; import org.springframework.cloud.stream.binder.ProducerProperties; import org.springframework.cloud.stream.binding.BinderAwareChannelResolver; +import org.springframework.cloud.stream.binding.BindingService; import org.springframework.cloud.stream.config.BindingProperties; import org.springframework.cloud.stream.config.BindingServiceProperties; import org.springframework.cloud.stream.messaging.DirectWithAttributesChannel; import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.annotation.Bean; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.SubscribableChannel; +import org.springframework.messaging.support.GenericMessage; +import org.springframework.stereotype.Component; import org.springframework.util.MimeType; import org.springframework.util.MimeTypeUtils; @@ -77,6 +90,9 @@ public final class StreamBridge implements SmartInitializingSingleton { @Autowired private BinderAwareChannelResolver dynamicDestinationResolver; + @Autowired + private BindingService bindingService; + /** * * @param functionCatalog instance of {@link FunctionCatalog} @@ -103,6 +119,10 @@ public final class StreamBridge implements SmartInitializingSingleton { return this.send(bindingName, data, MimeTypeUtils.APPLICATION_JSON); } + public boolean sendLight(String bindingName, Object data) { + return this.sendLight(bindingName, data, MimeTypeUtils.APPLICATION_JSON); + } + /** * Sends 'data' to an output binding specified by 'bindingName' argument while * using the content type specified by the 'outputContentType' argument to deal @@ -142,6 +162,26 @@ public final class StreamBridge implements SmartInitializingSingleton { return true; } + private Map channelCache = new HashMap<>(); + + @SuppressWarnings("unchecked") + public boolean sendLight(String bindingName, Object data, MimeType outputContentType) { + SubscribableChannel messageChannel = null; + if (channelCache.containsKey(bindingName)) { + messageChannel = channelCache.get(bindingName); + } + else { + messageChannel = new DirectWithAttributesChannel(); + Binding binding = this.bindingService.bindProducer(messageChannel, bindingName, false); + + ProducerProperties producerProperties = this.bindingServiceProperties.getProducerProperties(bindingName); + producerProperties.setRequiredGroups(bindingName); + channelCache.put(bindingName, messageChannel); + } + + return messageChannel.send(new GenericMessage<>(data)); + } + @Override public void afterSingletonsInstantiated() { if (this.initialized) {