diff --git a/docs/src/main/asciidoc/_configprops.adoc b/docs/src/main/asciidoc/_configprops.adoc index 99f09182b..fe3533b31 100644 --- a/docs/src/main/asciidoc/_configprops.adoc +++ b/docs/src/main/asciidoc/_configprops.adoc @@ -9,7 +9,6 @@ |spring.cloud.stream.dynamic-destinations | `[]` | A list of destinations that can be bound dynamically. If set, only listed destinations can be bound. |spring.cloud.stream.function.batch-mode | `false` | |spring.cloud.stream.function.bindings | | -|spring.cloud.stream.function.definition | | Definition of functions to bind. If several functions need to be composed into one, use pipes (e.g., 'fooFunc\|barFunc') |spring.cloud.stream.instance-count | `1` | The number of deployed instances of an application. Default: 1. NOTE: Could also be managed per individual binding "spring.cloud.stream.bindings.foo.consumer.instance-count" where 'foo' is the name of the binding. |spring.cloud.stream.instance-index | `0` | The instance id of the application: a number from 0 to instanceCount-1. Used for partitioning and with Kafka. NOTE: Could also be managed per individual binding "spring.cloud.stream.bindings.foo.consumer.instance-index" where 'foo' is the name of the binding. |spring.cloud.stream.instance-index-list | | A list of instance id's from 0 to instanceCount-1. Used for partitioning and with Kafka. NOTE: Could also be managed per individual binding "spring.cloud.stream.bindings.foo.consumer.instance-index-list" where 'foo' is the name of the binding. This setting will override the one set in 'spring.cloud.stream.instance-index' diff --git a/docs/src/main/asciidoc/spring-cloud-stream.adoc b/docs/src/main/asciidoc/spring-cloud-stream.adoc index b1e345236..b99e7455b 100644 --- a/docs/src/main/asciidoc/spring-cloud-stream.adoc +++ b/docs/src/main/asciidoc/spring-cloud-stream.adoc @@ -683,7 +683,6 @@ public class WebSourceApplication { @RequestMapping @ResponseStatus(HttpStatus.ACCEPTED) public void delegateToSupplier(@RequestBody String body) { - //processor.onNext(body); streamBridge.send("myBinidng", body); } } @@ -698,6 +697,32 @@ curl -H "Content-Type: text/plain" -X POST -d "hello from the other side" http:/ By showing two example we want to emphasize the approach will work with any type of foreign sources. +====== Output Content Type with StreamBridge + +You can also provide specific content type if necessary with the following method signature `public boolean send(String bindingName, Object data, MimeType outputContentType)`. +Or if you send data as a `Message`, its content type will be honored. + +====== Using specific binder type with StreamBridge + +Spring Cloud Stream supports multiple binder scenarios. For example you may be receiving data from Kafka and sending it to RabbitMQ. + +For more information on multiple binders scenarios, please see <> section and specifically <> + +In the event you are planning to use StreamBridge and have more then one binder configured in your application you must also tell StreamBridge +which binder to use. And for that there are two more variations of `send` method: + +[source, java] +---- +public boolean send(String bindingName, @Nullable String binderType, Object data) + +public boolean send(String bindingName, @Nullable String binderType, Object data, MimeType outputContentType) +---- + +As you can see there is one additional argument that you can provide - `binderType`, telling BindingService which binder to use when creating dynamic binding. + +NOTE: For cases where `spring.cloud.stream.source` property is used or the binding was already created under different binder, the `binderType` +argument will have no effect. + ===== Reactive Functions support 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 e8987f108..d06eaa9ff 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 @@ -1,5 +1,5 @@ /* - * Copyright 2015-2019 the original author or authors. + * Copyright 2015-2021 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -43,6 +43,7 @@ import org.springframework.cloud.stream.binder.PollableConsumerBinder; import org.springframework.cloud.stream.binder.PollableSource; import org.springframework.cloud.stream.binder.ProducerProperties; import org.springframework.cloud.stream.config.BindingServiceProperties; +import org.springframework.lang.Nullable; import org.springframework.scheduling.TaskScheduler; import org.springframework.util.CollectionUtils; import org.springframework.util.StringUtils; @@ -256,17 +257,19 @@ public class BindingService { } }); } + @SuppressWarnings({ "unchecked", "rawtypes" }) - public Binding bindProducer(T output, String outputName, boolean cache) { - String bindingTarget = this.bindingServiceProperties - .getBindingDestination(outputName); + public Binding bindProducer(T output, String outputName, boolean cache, @Nullable Binder binder) { + String bindingTarget = this.bindingServiceProperties.getBindingDestination(outputName); Class outputClass = output.getClass(); if (output instanceof Advised) { outputClass = Stream.of(((Advised) output).getProxiedInterfaces()).filter(c -> !c.getName().contains("org.springframework")).findFirst() .orElse(outputClass); } - Binder binder = (Binder) getBinder( - outputName, outputClass); + if (binder == null) { + binder = (Binder) getBinder(outputName, outputClass); + } + ProducerProperties producerProperties = this.bindingServiceProperties .getProducerProperties(outputName); if (binder instanceof ExtendedPropertiesBinder) { @@ -287,6 +290,10 @@ public class BindingService { return binding; } + public Binding bindProducer(T output, String outputName, boolean cache) { + return this.bindProducer(output, outputName, cache, null); + } + public Binding bindProducer(T output, String outputName) { return this.bindProducer(output, outputName, true); } 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 f5b8a6b99..61d677e88 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 @@ -233,7 +233,7 @@ public class FunctionConfiguration { .route(Message.class, message -> { if (message.getHeaders().get("spring.cloud.stream.sendto.destination") != null) { String destinationName = (String) message.getHeaders().get("spring.cloud.stream.sendto.destination"); - return streamBridge.resolveDestination(destinationName, producerProperties); + return streamBridge.resolveDestination(destinationName, producerProperties, null); } return outputName; }).get(); @@ -250,7 +250,7 @@ public class FunctionConfiguration { .route(Message.class, message -> { if (message.getHeaders().get("spring.cloud.stream.sendto.destination") != null) { String destinationName = (String) message.getHeaders().get("spring.cloud.stream.sendto.destination"); - return streamBridge.resolveDestination(destinationName, producerProperties); + return streamBridge.resolveDestination(destinationName, producerProperties, null); } return outputName; }) @@ -503,7 +503,7 @@ public class FunctionConfiguration { if (message instanceof Message && ((Message) message).getHeaders().get("spring.cloud.stream.sendto.destination") != null) { String destinationName = (String) ((Message) message).getHeaders().get("spring.cloud.stream.sendto.destination"); ProducerProperties producerProperties = this.serviceProperties.getBindings().get(outputBindingNames.iterator().next()).getProducer(); - MessageChannel dynamicChannel = streamBridge.resolveDestination(destinationName, producerProperties); + MessageChannel dynamicChannel = streamBridge.resolveDestination(destinationName, producerProperties, null); if (logger.isInfoEnabled()) { logger.info("Output message is sent to '" + destinationName + "' destination"); } @@ -580,7 +580,7 @@ public class FunctionConfiguration { private void doSendMessage(Object result, Message requestMessage) { if (result instanceof Message && ((Message) result).getHeaders().get("spring.cloud.stream.sendto.destination") != null) { String destinationName = (String) ((Message) result).getHeaders().get("spring.cloud.stream.sendto.destination"); - SubscribableChannel outputChannel = streamBridge.resolveDestination(destinationName, producerProperties); + SubscribableChannel outputChannel = streamBridge.resolveDestination(destinationName, producerProperties, null); if (logger.isInfoEnabled()) { logger.info("Output message is sent to '" + destinationName + "' destination"); } 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 90004cb08..19ecacd80 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 @@ -1,5 +1,5 @@ /* - * Copyright 2020-2020 the original author or authors. + * Copyright 2020-2021 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -26,12 +26,13 @@ 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.SimpleFunctionRegistry.FunctionInvocationWrapper; +import org.springframework.cloud.stream.binder.Binder; +import org.springframework.cloud.stream.binder.BinderFactory; import org.springframework.cloud.stream.binder.ProducerProperties; import org.springframework.cloud.stream.binding.BinderAwareChannelResolver.NewDestinationBindingCallback; import org.springframework.cloud.stream.binding.BindingService; @@ -47,6 +48,7 @@ import org.springframework.messaging.support.ChannelInterceptor; import org.springframework.util.CollectionUtils; import org.springframework.util.MimeType; import org.springframework.util.MimeTypeUtils; +import org.springframework.util.StringUtils; /** * A class which allows user to send data to an output binding. @@ -85,9 +87,7 @@ public final class StreamBridge implements SmartInitializingSingleton { private boolean initialized; - - @Autowired - private BindingService bindingService; + private final BindingService bindingService; /** * @@ -100,6 +100,7 @@ public final class StreamBridge implements SmartInitializingSingleton { StreamBridge(FunctionCatalog functionCatalog, FunctionRegistry functionRegistry, BindingServiceProperties bindingServiceProperties, ConfigurableApplicationContext applicationContext, @Nullable NewDestinationBindingCallback destinationBindingCallback) { + this.bindingService = applicationContext.getBean(BindingService.class); this.functionCatalog = functionCatalog; this.functionRegistry = functionRegistry; this.applicationContext = applicationContext; @@ -120,6 +121,11 @@ public final class StreamBridge implements SmartInitializingSingleton { /** * Sends 'data' to an output binding specified by 'bindingName' argument while * using default content type 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 truly dynamic destinations. This means if the name + * provided via 'bindingName' does not have a corresponding binding such name will be + * treated as dynamic destination.
+ * Will use default binder. For specific binder type see {@link #send(String, String, Object)} and {@link #send(String, String, Object, MimeType)} methods. * @param bindingName the name of the output binding. That said it requires a bit of clarification. * When using bridge.send("foo"...), the 'foo' typically represents the binding name. However * if such binding does not exist, the new binding will be created to support dynamic destinations. @@ -137,8 +143,8 @@ public final class StreamBridge implements SmartInitializingSingleton { * For typical cases `bindingName` is configured using 'spring.cloud.stream.source' property. * However, this operation also supports sending to truly dynamic destinations. This means if the name * provided via 'bindingName' does not have a corresponding binding such name will be - * treated as dynamic destination. - * + * treated as dynamic destination.
+ * Will use default binder. For specific binder type see {@link #send(String, String, Object)} and {@link #send(String, String, Object, MimeType)} methods. * @param bindingName the name of the output binding. That said it requires a bit of clarification. * When using bridge.send("foo"...), the 'foo' typically represents the binding name. However * if such binding does not exist, the new binding will be created to support dynamic destinations. @@ -146,13 +152,54 @@ public final class StreamBridge implements SmartInitializingSingleton { * @param outputContentType content type to be used to deal with output type conversion * @return true if data was sent successfully, otherwise false or throws an exception. */ - @SuppressWarnings("unchecked") public boolean send(String bindingName, Object data, MimeType outputContentType) { + return this.send(bindingName, null, data, outputContentType); + } + + /** + * 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 truly 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. That said it requires a bit of clarification. + * When using bridge.send("foo"...), the 'foo' typically represents the binding name. However + * if such binding does not exist, the new binding will be created to support dynamic destinations. + * @param binderType the type of the binder to use (e.g., 'kafka', 'rabbit') for cases where multiple binders are used. Can be null. + * @param data the data to send + * @return true if data was sent successfully, otherwise false or throws an exception. + */ + public boolean send(String bindingName, @Nullable String binderType, Object data) { + return this.send(bindingName, binderType, 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 + * 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 truly 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. That said it requires a bit of clarification. + * When using bridge.send("foo"...), the 'foo' typically represents the binding name. However + * if such binding does not exist, the new binding will be created to support dynamic destinations. + * @param binderType the type of the binder to use (e.g., 'kafka', 'rabbit') for cases where multiple binders are used. Can be null. + * @param data the data to send + * @param outputContentType content type to be used to deal with output type conversion + * @return true if data was sent successfully, otherwise false or throws an exception. + */ + @SuppressWarnings("unchecked") + public boolean send(String bindingName, @Nullable String binderType, Object data, MimeType outputContentType) { if (!(data instanceof Message)) { data = MessageBuilder.withPayload(data).build(); } ProducerProperties producerProperties = this.bindingServiceProperties.getProducerProperties(bindingName); - SubscribableChannel messageChannel = this.resolveDestination(bindingName, producerProperties); + SubscribableChannel messageChannel = this.resolveDestination(bindingName, producerProperties, binderType); boolean skipConversion = producerProperties.isUseNativeEncoding(); @@ -184,8 +231,8 @@ public final class StreamBridge implements SmartInitializingSingleton { this.initialized = true; } - @SuppressWarnings("unchecked") - synchronized SubscribableChannel resolveDestination(String destinationName, ProducerProperties producerProperties) { + @SuppressWarnings({ "unchecked", "rawtypes"}) + synchronized SubscribableChannel resolveDestination(String destinationName, ProducerProperties producerProperties, String binderName) { SubscribableChannel messageChannel = this.channelCache.get(destinationName); if (messageChannel == null && this.applicationContext.containsBean(destinationName)) { messageChannel = this.applicationContext.getBean(destinationName, SubscribableChannel.class); @@ -200,7 +247,13 @@ public final class StreamBridge implements SmartInitializingSingleton { producerProperties, extendedProducerProperties); } - this.bindingService.bindProducer(messageChannel, destinationName, false); + Binder binder = null; + if (StringUtils.hasText(binderName)) { + BinderFactory binderFactory = this.applicationContext.getBean(BinderFactory.class); + binder = binderFactory.getBinder(binderName, messageChannel.getClass()); + } + + this.bindingService.bindProducer(messageChannel, destinationName, false, binder); this.channelCache.put(destinationName, messageChannel); this.addInterceptors((AbstractMessageChannel) messageChannel); }