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.

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java

@@ -249,9 +249,8 @@ public class BindingService {
 			}
 		});
 	}
-
 	@SuppressWarnings({ "unchecked", "rawtypes" })
-	public <T> Binding<T> bindProducer(T output, String outputName) {
+	public <T> Binding<T> bindProducer(T output, String outputName, boolean cache) {
 		String bindingTarget = this.bindingServiceProperties
 				.getBindingDestination(outputName);
 		Binder<T, ?, ProducerProperties> binder = (Binder<T, ?, ProducerProperties>) getBinder(
@@ -270,10 +269,16 @@ public class BindingService {
 		validate(producerProperties);
 		Binding<T> binding = doBindProducer(output, bindingTarget, binder,
 				producerProperties);
-		this.producerBindings.put(outputName, binding);
+		if (cache) {
+			this.producerBindings.put(outputName, binding);
+		}
 		return binding;
 	}
 
+	public <T> Binding<T> 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());

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<String, SubscribableChannel> 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<SubscribableChannel> 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) {