From 8a30fea67a429f2a42b099f3ceec3ec5d8b6df53 Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Tue, 24 Mar 2020 17:09:07 +0100 Subject: [PATCH] GH-1924 Add support for dynamic destinations to StramBridge Resolves #1924 --- .../main/asciidoc/spring-cloud-stream.adoc | 36 +++++++++++++++++++ .../function/FunctionConfiguration.java | 5 ++- .../cloud/stream/function/StreamBridge.java | 29 +++++++++++++-- .../stream/function/StreamBridgeTests.java | 26 +++++++++++++- 4 files changed, 89 insertions(+), 7 deletions(-) diff --git a/docs/src/main/asciidoc/spring-cloud-stream.adoc b/docs/src/main/asciidoc/spring-cloud-stream.adoc index 1a9682ae4..1c5f78eb5 100644 --- a/docs/src/main/asciidoc/spring-cloud-stream.adoc +++ b/docs/src/main/asciidoc/spring-cloud-stream.adoc @@ -661,6 +661,39 @@ Also, note that `streamBridge.send(..)` method takes an `Object` for data. This will go through the same routine when sending output as if it was from any Function or Supplier providing the same level of consistency as with functions. This means the output type conversion, partitioning etc are honored as if it was from the output produced by functions. +====== StreamBridge and Dynamic Destinations + +`StreamBridge` can also be used for cases when output destination(s) are not known ahead of time similar to the use cases +described in <> section. + +Let's look at the example + +[source, java] +---- +@SpringBootApplication +@Controller +public class WebSourceApplication { + + public static void main(String[] args) { + SpringApplication.run(WebSourceApplication.class, args); + } + + @Autowired + private StreamBridge streamBridge; + + @RequestMapping + @ResponseStatus(HttpStatus.ACCEPTED) + public void delegateToSupplier(@RequestBody String body) { + System.out.println("Sending " + body); + streamBridge.send("myDestiniation", body); + } +} +---- + +As you can see the preceding example is very similar to the previous one with the exception of explicit binding instruction provided via +`spring.cloud.stream.source` property (which is not provided). +Here we're sending data to `myDestiniation` name which does not exist as a binding. Therefore such name will be treated as dynamic destination +as described in <> section. ====== Using reactor API @@ -1472,6 +1505,9 @@ public NewDestinationBindingCallback dynamicConfigurer NOTE: If you need to support dynamic destinations with multiple binder types, use `Object` for the generic type and cast the `extended` argument as needed. +Also, please see <> section to see how yet another option (StreamBridge) can be utilized for similar cases. + + [[spring-cloud-stream-overview-error-handling]] === Error Handling 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 210132a41..dbdf6ed56 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 @@ -48,7 +48,6 @@ import org.springframework.beans.factory.support.RootBeanDefinition; import org.springframework.boot.autoconfigure.AutoConfigureAfter; import org.springframework.boot.autoconfigure.AutoConfigureBefore; import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; -import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.cloud.function.context.FunctionCatalog; import org.springframework.cloud.function.context.FunctionProperties; @@ -122,7 +121,7 @@ public class FunctionConfiguration { private final static String SOURCE_PROPERY = "spring.cloud.stream.source"; @Bean - @ConditionalOnProperty(SOURCE_PROPERY) +// @ConditionalOnProperty(SOURCE_PROPERY) public StreamBridge streamBridgeUtils(FunctionCatalog functionCatalog, FunctionRegistry functionRegistry, BindingServiceProperties bindingServiceProperties, ConfigurableApplicationContext applicationContext) { return new StreamBridge(functionCatalog, functionRegistry, bindingServiceProperties, applicationContext); @@ -188,7 +187,7 @@ public class FunctionConfiguration { String integrationFlowName = proxyFactory.getFunctionDefinition() + "_integrationflow"; PollableBean pollable = extractPollableAnnotation(functionProperties, context, proxyFactory); - Type functionType = ((FunctionInvocationWrapper) functionWrapper).getFunctionType(); + Type functionType = functionWrapper.getFunctionType(); IntegrationFlow integrationFlow = integrationFlowFromProvidedSupplier(new PartitionAwareFunctionWrapper(functionWrapper, context, producerProperties), beginPublishingTrigger, pollable, context, taskScheduler, functionType) .route(Message.class, message -> { 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 7e277986f..943b05e4a 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 @@ -21,19 +21,24 @@ import java.util.Map; import java.util.Map.Entry; import java.util.function.Function; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + import org.springframework.beans.factory.SmartInitializingSingleton; +import org.springframework.beans.factory.annotation.Autowired; 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.BeanFactoryAwareFunctionRegistry.FunctionInvocationWrapper; import org.springframework.cloud.stream.binder.ProducerProperties; +import org.springframework.cloud.stream.binding.BinderAwareChannelResolver; 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.messaging.Message; -import org.springframework.util.Assert; +import org.springframework.messaging.MessageChannel; import org.springframework.util.MimeType; import org.springframework.util.MimeTypeUtils; @@ -55,7 +60,9 @@ import org.springframework.util.MimeTypeUtils; */ public final class StreamBridge implements SmartInitializingSingleton { - private final Map outputChannelsOnly = new HashMap<>(); + protected final Log logger = LogFactory.getLog(getClass()); + + private final Map outputChannelsOnly = new HashMap<>(); private final FunctionCatalog functionCatalog; @@ -67,6 +74,9 @@ public final class StreamBridge implements SmartInitializingSingleton { private boolean initialized; + @Autowired + private BinderAwareChannelResolver dynamicDestinationResolver; + /** * * @param functionCatalog instance of {@link FunctionCatalog} @@ -97,6 +107,11 @@ public final class StreamBridge implements SmartInitializingSingleton { * Sends 'data' to an output binding specified by 'bindingName' argument while * using the content type specified by the 'outputContentType' argument to deal * with output type conversion (if necessary). + * For typical cases `bindingName` is configured using 'spring.cloud.stream.source' property. + * However, this operation also supports sending to dynamic destinations. This means if the name + * provided via 'bindingName' does not have a corresponding binding such name will be + * treated as dynamic destination. + * * @param bindingName the name of the output binding * @param data the data to send * @param outputContentType content type to be used to deal with output type conversion @@ -104,7 +119,15 @@ public final class StreamBridge implements SmartInitializingSingleton { */ @SuppressWarnings("unchecked") public boolean send(String bindingName, Object data, MimeType outputContentType) { - Assert.isTrue(this.outputChannelsOnly.containsKey(bindingName), "Binding name '" + bindingName + "' does not exist."); + if (!this.outputChannelsOnly.containsKey(bindingName)) { + logger.info("Binding name '" + bindingName + "' does not exist. This means that value '" + + bindingName + "' will be treated as dynamic destination. If this is not your intention please " + + "provide 'spring.cloud.stream.source' property"); + this.outputChannelsOnly.put(bindingName, dynamicDestinationResolver.resolveDestination(bindingName)); + FunctionRegistration> fr = new FunctionRegistration<>(v -> v, bindingName); + this.functionRegistry.register(fr.type(FunctionType.from(Object.class).to(Object.class).message())); + } + FunctionInvocationWrapper functionWrapper = this.functionCatalog.lookup(bindingName, outputContentType.toString()); BindingProperties bindingProperties = this.bindingServiceProperties.getBindings().get(bindingName); diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/StreamBridgeTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/StreamBridgeTests.java index 0e7d1cdca..c83f9fa6e 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/StreamBridgeTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/StreamBridgeTests.java @@ -92,7 +92,7 @@ public class StreamBridgeTests { OutputDestination outputDestination = context.getBean(OutputDestination.class); - Message message = outputDestination.receive(100, "foo-out-0"); + Message message = outputDestination.receive(100, "foo-out-0"); assertThat(message.getPayload()).isEqualTo("hello foo".getBytes()); assertThat(message.getHeaders().get("foo")).isEqualTo("foo"); @@ -146,6 +146,30 @@ public class StreamBridgeTests { } } + @Test + public void testDynamicDestination() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder(TestChannelBinderConfiguration + .getCompleteConfiguration(TestConfiguration.class)) + .web(WebApplicationType.NONE).run("--spring.jmx.enabled=false")) { + + OutputDestination outputDestination = context.getBean(OutputDestination.class); + + StreamBridge bridge = context.getBean(StreamBridge.class); + bridge.send("foo-out-0", "b"); + bridge.send("bar", "hello"); + bridge.send("blah", MessageBuilder.withPayload("message").setHeader("foo", "foo").build()); + + Message message = outputDestination.receive(100, "foo-out-0"); + assertThat(new String(message.getPayload())).isEqualTo("b"); + + message = outputDestination.receive(100, "bar"); + assertThat(new String(message.getPayload())).isEqualTo("hello"); + + message = outputDestination.receive(100, "blah"); + assertThat(new String(message.getPayload())).isEqualTo("message"); + } + } + @EnableAutoConfiguration public static class EmptyConfiguration {