SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-2114 Add support for multiple binders to StreamBridge

Browse Source 
Resolves #2114
This commit is contained in:
Oleg Zhurakousky
2021-05-25 14:33:25 +02:00
parent 717a8e4f8b
commit 2677035779
5 changed files with 108 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
docs/src/main/asciidoc/_configprops.adoc
View File

@@ -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'

27
docs/src/main/asciidoc/spring-cloud-stream.adoc
View File

@@ -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 <<Binders>> section and specifically <<Multiple Binders on the Classpath>>
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

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

@@ -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 <T> Binding<T> bindProducer(T output, String outputName, boolean cache) {
String bindingTarget = this.bindingServiceProperties
.getBindingDestination(outputName);
public <T> Binding<T> bindProducer(T output, String outputName, boolean cache, @Nullable Binder<T, ?, ProducerProperties> 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<T, ?, ProducerProperties> binder = (Binder<T, ?, ProducerProperties>) getBinder(
outputName, outputClass);
if (binder == null) {
binder = (Binder<T, ?, ProducerProperties>) getBinder(outputName, outputClass);
}
ProducerProperties producerProperties = this.bindingServiceProperties
.getProducerProperties(outputName);
if (binder instanceof ExtendedPropertiesBinder) {
@@ -287,6 +290,10 @@ public class BindingService {
return binding;
}
public <T> Binding<T> bindProducer(T output, String outputName, boolean cache) {
return this.bindProducer(output, outputName, cache, null);
}
public <T> Binding<T> bindProducer(T output, String outputName) {
return this.bindProducer(output, outputName, true);
}

8
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java
View File

@@ -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");
}

77
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java
View File

@@ -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.<br>
* 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.<br>
* 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);
}