From df289bc65dd25ec6d205d38842686179cc13c1db Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Thu, 19 Jan 2023 10:29:05 +0100 Subject: [PATCH] GH-2626 Ensure explicit binding names are not modified See issue for more details Resolves #2626 --- .../stream/binding/BindingServiceTests.java | 4 +- .../stream/binding/ExplicitBindingTests.java | 8 +-- .../stream/function/StreamBridgeTests.java | 50 +++++++++---------- .../BindableFunctionProxyFactory.java | 6 +++ .../function/FunctionConfiguration.java | 2 +- .../CustomPartitionedProducerTest.java | 14 +++--- 6 files changed, 45 insertions(+), 39 deletions(-) diff --git a/core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/binding/BindingServiceTests.java b/core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/binding/BindingServiceTests.java index e301e72f7..fa0abd885 100644 --- a/core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/binding/BindingServiceTests.java +++ b/core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/binding/BindingServiceTests.java @@ -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"); diff --git a/core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/binding/ExplicitBindingTests.java b/core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/binding/ExplicitBindingTests.java index bb6bcd543..e1dc533cd 100644 --- a/core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/binding/ExplicitBindingTests.java +++ b/core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/binding/ExplicitBindingTests.java @@ -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(); } } diff --git a/core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/function/StreamBridgeTests.java b/core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/function/StreamBridgeTests.java index 3d1b28f4e..ffaa2d0ea 100644 --- a/core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/function/StreamBridgeTests.java +++ b/core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/function/StreamBridgeTests.java @@ -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 message = outputDestination.receive(100, "foo-out-0"); + Message 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); diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/BindableFunctionProxyFactory.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/BindableFunctionProxyFactory.java index 9f7f3c023..077bd5f8e 100644 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/BindableFunctionProxyFactory.java +++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/BindableFunctionProxyFactory.java @@ -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) diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java index bfb624708..91d996929 100644 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java +++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java @@ -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); } diff --git a/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/CustomPartitionedProducerTest.java b/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/CustomPartitionedProducerTest.java index f1716f0af..b871f9bd8 100644 --- a/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/CustomPartitionedProducerTest.java +++ b/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/CustomPartitionedProducerTest.java @@ -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 timerMessageSource() { return new MessageSource() { @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 timerMessageSource() { return new MessageSource() { @Override