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

Set 'setRequiredGroups' in StreamBridge

Browse Source 
This is set b y default for trully dynamic cases which are cases where source bindings such as 'spring.cloud.stream.source' were not created
This commit is contained in:
Oleg Zhurakousky
2020-05-11 20:40:48 +02:00
parent b3988f52aa
commit 8008ccaab9
1 changed files with 2 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -131,7 +131,9 @@ public final class StreamBridge implements SmartInitializingSingleton {
@SuppressWarnings({ "unchecked", "unused" })
public boolean send(String bindingName, Object data, MimeType outputContentType) {
SubscribableChannel messageChannel = this.channelCache.get(bindingName);
ProducerProperties producerProperties = this.bindingServiceProperties.getProducerProperties(bindingName);
if (messageChannel == null) {
producerProperties.setRequiredGroups(bindingName);
FunctionRegistration<Function<Object, Object>> fr = new FunctionRegistration<>(v -> v, bindingName);
this.functionRegistry.register(fr.type(FunctionType.from(Object.class).to(Object.class).message()));
messageChannel = new DirectWithAttributesChannel();
@@ -139,8 +141,6 @@ public final class StreamBridge implements SmartInitializingSingleton {
this.channelCache.put(bindingName, messageChannel);
}
ProducerProperties producerProperties = this.bindingServiceProperties.getProducerProperties(bindingName);
//producerProperties.setRequiredGroups(bindingName);
Function<Object, Object> functionToInvoke = this.functionCatalog.lookup(bindingName, outputContentType.toString());
if (producerProperties != null && producerProperties.isPartitioned()) {
functionToInvoke = new PartitionAwareFunctionWrapper((FunctionInvocationWrapper) functionToInvoke, this.applicationContext, producerProperties);