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

GH-2626 Ensure explicit binding names are not modified

Browse Source 
See issue for more details
Resolves #2626
This commit is contained in:
Oleg Zhurakousky
2023-01-19 10:29:05 +01:00
parent e8a2018b08
commit df289bc65d
6 changed files with 45 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/binding/BindingServiceTests.java
View File

@@ -366,7 +366,7 @@ public class BindingServiceTests {
assertThat(bindings.get("input2").getConsumer().getConcurrency()).isEqualTo(1);
assertThat(bindings.get("input1").getConsumer().isPartitioned()).isEqualTo(true);
assertThat(bindings.get("input2").getConsumer().isPartitioned()).isEqualTo(false);
assertThat(bindings.get("output1-out-0").getProducer().getPartitionCount())
assertThat(bindings.get("output1").getProducer().getPartitionCount())
.isEqualTo(10);
assertThat(bindings.get("output2").getProducer().getPartitionCount())
.isEqualTo(1);
@@ -375,7 +375,7 @@ public class BindingServiceTests {
.isEqualTo("application/json");
assertThat(bindings.get("inputFooBar").getContentType())
.isEqualTo("application/avro");
assertThat(bindings.get("inputFooBarBuzz-in-0").getContentType())
assertThat(bindings.get("inputFooBarBuzz").getContentType())
.isEqualTo("text/plain");
assertThat(bindings.get("input_snake_case").getContentType())
.isEqualTo("application/avro");

8
core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/binding/ExplicitBindingTests.java
View File

@@ -49,10 +49,10 @@ public class ExplicitBindingTests {
"--spring.cloud.stream.input-bindings=fooin;barin",
"--spring.cloud.stream.output-bindings=fooout;barout")) {
assertThat(context.getBean("fooin-in-0", MessageChannel.class)).isNotNull();
assertThat(context.getBean("barin-in-0", MessageChannel.class)).isNotNull();
assertThat(context.getBean("fooout-out-0", MessageChannel.class)).isNotNull();
assertThat(context.getBean("barout-out-0", MessageChannel.class)).isNotNull();
assertThat(context.getBean("fooin", MessageChannel.class)).isNotNull();
assertThat(context.getBean("barin", MessageChannel.class)).isNotNull();
assertThat(context.getBean("fooout", MessageChannel.class)).isNotNull();
assertThat(context.getBean("barout", MessageChannel.class)).isNotNull();
}
}

50
core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/function/StreamBridgeTests.java
View File

@@ -319,18 +319,16 @@ public class StreamBridgeTests {
.web(WebApplicationType.NONE).run(
"--spring.jmx.enabled=false",
"--spring.cloud.stream.dynamic-destination-cache-size=1",
"--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"
"--spring.cloud.stream.output-bindings=outputA;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");
bridge.send("outputA", "hello foo");
bridge.send("outputA", "hello foo");
bridge.send("outputA", "hello foo");
bridge.send("outputA", "hello foo");
AbstractMessageChannel messageChannel = context.getBean("outputA-out-0", AbstractMessageChannel.class);
AbstractMessageChannel messageChannel = context.getBean("outputA", AbstractMessageChannel.class);
assertThat(messageChannel.getInterceptors()).hasSize(1);
}
@@ -479,49 +477,49 @@ public class StreamBridgeTests {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(TestChannelBinderConfiguration
.getCompleteConfiguration(EmptyConfiguration.class))
.web(WebApplicationType.NONE).run("--spring.cloud.stream.source=foo;bar",
"--spring.cloud.stream.bindings.foo-out-0.producer.partitionKeyExpression=payload",
"--spring.cloud.stream.bindings.foo-out-0.producer.partitionCount=5",
"--spring.cloud.stream.bindings.bar-out-0.producer.partitionKeyExpression=payload",
"--spring.cloud.stream.bindings.bar-out-0.producer.partitionCount=1",
"--spring.cloud.stream.bindings.foo.producer.partitionKeyExpression=payload",
"--spring.cloud.stream.bindings.foo.producer.partitionCount=5",
"--spring.cloud.stream.bindings.bar.producer.partitionKeyExpression=payload",
"--spring.cloud.stream.bindings.bar.producer.partitionCount=1",
"--spring.jmx.enabled=false")) {
StreamBridge bridge = context.getBean(StreamBridge.class);
bridge.send("foo-out-0", "a");
bridge.send("bar-out-0", "b");
bridge.send("foo-out-0", "c");
bridge.send("foo-out-0", "d");
bridge.send("bar-out-0", "e");
bridge.send("foo-out-0", "f");
bridge.send("bar-out-0", "g");
bridge.send("foo", "a");
bridge.send("bar", "b");
bridge.send("foo", "c");
bridge.send("foo", "d");
bridge.send("bar", "e");
bridge.send("foo", "f");
bridge.send("bar", "g");
OutputDestination outputDestination = context.getBean(OutputDestination.class);
Message<byte[]> message = outputDestination.receive(100, "foo-out-0");
Message<byte[]> message = outputDestination.receive(100, "foo");
assertThat(new String(message.getPayload())).isEqualTo("a");
assertThat(message.getHeaders().get("scst_partition")).isEqualTo(2);
message = outputDestination.receive(100, "foo-out-0");
message = outputDestination.receive(100, "foo");
assertThat(new String(message.getPayload())).isEqualTo("c");
assertThat(message.getHeaders().get("scst_partition")).isEqualTo(4);
message = outputDestination.receive(100, "foo-out-0");
message = outputDestination.receive(100, "foo");
assertThat(new String(message.getPayload())).isEqualTo("d");
assertThat(message.getHeaders().get("scst_partition")).isEqualTo(0);
message = outputDestination.receive(100, "bar-out-0");
message = outputDestination.receive(100, "bar");
assertThat(new String(message.getPayload())).isEqualTo("b");
assertThat(message.getHeaders().get("scst_partition")).isEqualTo(0);
message = outputDestination.receive(100, "bar-out-0");
message = outputDestination.receive(100, "bar");
assertThat(new String(message.getPayload())).isEqualTo("e");
assertThat(message.getHeaders().get("scst_partition")).isEqualTo(0);
message = outputDestination.receive(100, "bar-out-0");
message = outputDestination.receive(100, "bar");
assertThat(new String(message.getPayload())).isEqualTo("g");
assertThat(message.getHeaders().get("scst_partition")).isEqualTo(0);
message = outputDestination.receive(100, "foo-out-0");
message = outputDestination.receive(100, "foo");
assertThat(new String(message.getPayload())).isEqualTo("f");
assertThat(message.getHeaders().get("scst_partition")).isEqualTo(2);

6
core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/BindableFunctionProxyFactory.java
View File

@@ -139,6 +139,9 @@ public class BindableFunctionProxyFactory extends BindableProxyFactory implement
}
private String buildInputNameForIndex(int index) {
if (!this.isFunctionExist()) {
return this.functionDefinition;
}
return new StringBuilder(this.functionDefinition.replace(",", "|").replace("|", ""))
.append(FunctionConstants.DELIMITER)
.append(FunctionConstants.DEFAULT_INPUT_SUFFIX)
@@ -148,6 +151,9 @@ public class BindableFunctionProxyFactory extends BindableProxyFactory implement
}
private String buildOutputNameForIndex(int index) {
if (!this.isFunctionExist()) {
return this.functionDefinition;
}
return new StringBuilder(this.functionDefinition.replace(",", "|").replace("|", ""))
.append(FunctionConstants.DELIMITER)
.append(FunctionConstants.DEFAULT_OUTPUT_SUFFIX)

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

@@ -885,7 +885,7 @@ public class FunctionConfiguration {
sourceFunc.isSupplier() ||
(!sourceFunc.getFunctionDefinition().equals(inputBindingName) && applicationContext.containsBean(inputBindingName))) {
BindableFunctionProxyFactory proxyFactory = new BindableFunctionProxyFactory(inputBindingName, 1, 0, this.streamFunctionProperties);
BindableFunctionProxyFactory proxyFactory = new BindableFunctionProxyFactory(inputBindingName, 1, 0, this.streamFunctionProperties, sourceFunc != null);
((GenericApplicationContext) this.applicationContext).registerBean(inputBindingName + "_binding_in",
BindableFunctionProxyFactory.class, () -> proxyFactory);
}

14
core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/CustomPartitionedProducerTest.java
View File

@@ -58,7 +58,7 @@ public class CustomPartitionedProducerTest {
"--spring.cloud.stream.bindings.output.producer.partitionSelectorClass="
+ "org.springframework.cloud.stream.partitioning.CustomPartitionSelectorClass",
"--spring.cloud.stream.default-binder=mock");
DirectChannel messageChannel = context.getBean("output-out-0", DirectChannel.class);
DirectChannel messageChannel = context.getBean("output", DirectChannel.class);
for (ChannelInterceptor channelInterceptor : messageChannel
.getInterceptors()) {
if (channelInterceptor instanceof MessageConverterConfigurer.PartitioningInterceptor) {
@@ -93,7 +93,7 @@ public class CustomPartitionedProducerTest {
"--spring.cloud.stream.bindings.output.producer.partitionKeyExtractorName=customPartitionKeyExtractor",
"--spring.cloud.stream.bindings.output.producer.partitionSelectorName=customPartitionSelector",
"--spring.cloud.stream.default-binder=mock");
DirectChannel messageChannel = context.getBean("output-out-0", DirectChannel.class);
DirectChannel messageChannel = context.getBean("output", DirectChannel.class);
for (ChannelInterceptor channelInterceptor : messageChannel
.getInterceptors()) {
if (channelInterceptor instanceof MessageConverterConfigurer.PartitioningInterceptor) {
@@ -126,7 +126,7 @@ public class CustomPartitionedProducerTest {
"--spring.cloud.stream.output-bindings=output",
"--spring.jmx.enabled=false", "--spring.main.web-application-type=none",
"--spring.cloud.stream.default-binder=mock");
DirectChannel messageChannel = context.getBean("output-out-0", DirectChannel.class);
DirectChannel messageChannel = context.getBean("output", DirectChannel.class);
for (ChannelInterceptor channelInterceptor : messageChannel
.getInterceptors()) {
if (channelInterceptor instanceof MessageConverterConfigurer.PartitioningInterceptor) {
@@ -152,14 +152,16 @@ public class CustomPartitionedProducerTest {
}
}
@Test
public void testCustomPartitionedProducerMultipleInstances() {
ApplicationContext context = SpringApplication.run(
CustomPartitionedProducerTest.TestSourceMultipleStrategies.class,
"--spring.cloud.stream.output-bindings=output",
"--spring.jmx.enabled=false", "--spring.main.web-application-type=none",
"--spring.cloud.stream.bindings.output.producer.partitionKeyExtractorName=customPartitionKeyExtractorOne",
"--spring.cloud.stream.bindings.output.producer.partitionSelectorName=customPartitionSelectorTwo",
"--spring.cloud.stream.default-binder=mock");
DirectChannel messageChannel = context.getBean("output-out-0", DirectChannel.class);
DirectChannel messageChannel = context.getBean("output", DirectChannel.class);
for (ChannelInterceptor channelInterceptor : messageChannel
.getInterceptors()) {
if (channelInterceptor instanceof MessageConverterConfigurer.PartitioningInterceptor) {
@@ -200,7 +202,7 @@ public class CustomPartitionedProducerTest {
}
@Bean
@InboundChannelAdapter(value = "output-out-0", poller = @Poller(fixedDelay = "5000", maxMessagesPerPoll = "1"))
@InboundChannelAdapter(value = "output", poller = @Poller(fixedDelay = "5000", maxMessagesPerPoll = "1"))
public MessageSource<String> timerMessageSource() {
return new MessageSource<String>() {
@Override
@@ -238,7 +240,7 @@ public class CustomPartitionedProducerTest {
}
@Bean
@InboundChannelAdapter(value = "output-out-0", poller = @Poller(fixedDelay = "5000", maxMessagesPerPoll = "1"))
@InboundChannelAdapter(value = "output", poller = @Poller(fixedDelay = "5000", maxMessagesPerPoll = "1"))
public MessageSource<String> timerMessageSource() {
return new MessageSource<String>() {
@Override