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

GH-2266 Fix channel interceptor application for dynamic destinations

Browse Source 
Resolves #2266
Resolves #2269
This commit is contained in:
Oleg Zhurakousky
2022-01-18 10:32:12 +01:00
parent 36e93e2c6f
commit ff281c56c4
5 changed files with 11 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java
View File

@@ -147,7 +147,9 @@ public class MessageConverterConfigurer
ProducerProperties producerProperties = bindingProperties.getProducer();
boolean partitioned = !inbound && producerProperties != null && producerProperties.isPartitioned();
boolean functional = streamFunctionProperties != null
&& (StringUtils.hasText(streamFunctionProperties.getDefinition()) || StringUtils.hasText(bindingServiceProperties.getSource()));
&& (StringUtils.hasText(streamFunctionProperties.getDefinition())
|| StringUtils.hasText(bindingServiceProperties.getInputBindings())
|| StringUtils.hasText(bindingServiceProperties.getOutputBindings()));
if (partitioned) {
if (inbound || !functional) {

5
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceProperties.java
View File

@@ -44,6 +44,7 @@ import org.springframework.core.convert.converter.Converter;
import org.springframework.core.convert.support.GenericConversionService;
import org.springframework.integration.support.utils.IntegrationUtils;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
/**
* @author Dave Syer
@@ -330,6 +331,7 @@ public class BindingServiceProperties
@Deprecated
public void setSource(String source) {
this.source = source;
this.outputBindings = source;
}
public void updateProducerProperties(String bindingName,
@@ -356,10 +358,13 @@ public class BindingServiceProperties
}
public String getOutputBindings() {
return outputBindings;
}
public void setOutputBindings(String outputBindings) {
Assert.state(!StringUtils.hasText(this.source), "Setting 'source' and 'output-binding' is not allowed "
+ "because 'source' is deprecated in favor of 'output-binding'.");
this.outputBindings = outputBindings;
}

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

@@ -129,8 +129,6 @@ public class FunctionConfiguration {
private final static String SOURCE_PROPERY = "spring.cloud.stream.source";
// private final static String OUT_BINDINGS = "spring.cloud.stream.output-bindings";
@Bean
public StreamBridge streamBridgeUtils(FunctionCatalog functionCatalog, FunctionRegistry functionRegistry,
BindingServiceProperties bindingServiceProperties, ConfigurableApplicationContext applicationContext,

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

@@ -256,7 +256,6 @@ public final class StreamBridge implements SmartInitializingSingleton {
SubscribableChannel messageChannel = this.channelCache.get(destinationName);
if (messageChannel == null && this.applicationContext.containsBean(destinationName)) {
messageChannel = this.applicationContext.getBean(destinationName, SubscribableChannel.class);
this.addInterceptors((AbstractMessageChannel) messageChannel, destinationName);
}
if (messageChannel == null) {
messageChannel = new DirectWithAttributesChannel();

4
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/StreamBridgeTests.java
View File

@@ -193,11 +193,13 @@ public class StreamBridgeTests {
.web(WebApplicationType.NONE).run(
"--spring.jmx.enabled=false",
"--spring.cloud.stream.dynamic-destination-cache-size=1",
"--spring.cloud.stream.source=outputA;outputB",
"--spring.cloud.stream.output-bindings=outputA;outputB",
"--spring.cloud.stream.bindings.outputA-out-0.destination=outputA",
"--spring.cloud.stream.bindings.outputB-out-0.destination=outputB")) {
StreamBridge bridge = context.getBean(StreamBridge.class);
bridge.send("outputA-out-0", "hello foo");
bridge.send("outputA-out-0", "hello foo");
bridge.send("outputA-out-0", "hello foo");
bridge.send("outputA-out-0", "hello foo");