From 9b55bb838a6e529772afbcb52cb267f7ad2dffcb Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Mon, 22 Mar 2021 18:14:16 +0100 Subject: [PATCH] GH-2132 Fix channel creation/registration logic SubscribableChannelBindingTargetFactory now capable of dealing with channels that may have been created outside of the normal process. In other words if channel exists as bean it will only cofigure it Resolves #2132 --- ...bscribableChannelBindingTargetFactory.java | 61 ++++++++++++++----- .../ImplicitFunctionBindingTests.java | 33 ++++++++++ 2 files changed, 80 insertions(+), 14 deletions(-) diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/SubscribableChannelBindingTargetFactory.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/SubscribableChannelBindingTargetFactory.java index 2b301e711..93830eb72 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/SubscribableChannelBindingTargetFactory.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/SubscribableChannelBindingTargetFactory.java @@ -16,10 +16,9 @@ package org.springframework.cloud.stream.binding; +import org.springframework.beans.factory.BeanCreationException; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.cloud.stream.messaging.DirectWithAttributesChannel; -import org.springframework.cloud.stream.messaging.Sink; -import org.springframework.cloud.stream.messaging.Source; import org.springframework.context.support.GenericApplicationContext; import org.springframework.messaging.SubscribableChannel; @@ -48,25 +47,59 @@ public class SubscribableChannelBindingTargetFactory @Override public SubscribableChannel createInput(String name) { - DirectWithAttributesChannel subscribableChannel = new DirectWithAttributesChannel(); - subscribableChannel.setComponentName(name); - subscribableChannel.setAttribute("type", Sink.INPUT); - this.messageChannelConfigurer.configureInputChannel(subscribableChannel, name); - if (context != null && !context.containsBean(name)) { - context.registerBean(name, DirectWithAttributesChannel.class, () -> subscribableChannel); + SubscribableChannel subscribableChannel = null; + if (context != null && context.containsBean(name)) { + try { + subscribableChannel = context.getBean(name, SubscribableChannel.class); + } + catch (BeanCreationException e) { + // ignore + /* + * Since we still support annotation-based programming model, this exception happens + * because of proxies related to @Input @Output + */ + } } + if (subscribableChannel == null) { + DirectWithAttributesChannel channel = new DirectWithAttributesChannel(); + channel.setComponentName(name); + if (context != null && !context.containsBean(name)) { + context.registerBean(name, DirectWithAttributesChannel.class, () -> channel); + } + subscribableChannel = channel; + } + if (subscribableChannel instanceof DirectWithAttributesChannel) { + ((DirectWithAttributesChannel) subscribableChannel).setAttribute("type", "input"); + this.messageChannelConfigurer.configureInputChannel(subscribableChannel, name); + } + return subscribableChannel; } @Override public SubscribableChannel createOutput(String name) { - DirectWithAttributesChannel subscribableChannel = new DirectWithAttributesChannel(); - subscribableChannel.setComponentName(name); - subscribableChannel.setAttribute("type", Source.OUTPUT); - this.messageChannelConfigurer.configureOutputChannel(subscribableChannel, name); - if (context != null && !context.containsBean(name)) { - context.registerBean(name, DirectWithAttributesChannel.class, () -> subscribableChannel); + SubscribableChannel subscribableChannel = null; + if (context != null && context.containsBean(name)) { + try { + subscribableChannel = context.getBean(name, SubscribableChannel.class); + } + catch (BeanCreationException e) { + // ignore + } } + if (subscribableChannel == null) { + DirectWithAttributesChannel channel = new DirectWithAttributesChannel(); + channel.setComponentName(name); + if (context != null && !context.containsBean(name)) { + context.registerBean(name, DirectWithAttributesChannel.class, () -> channel); + } + subscribableChannel = channel; + } + if (subscribableChannel instanceof DirectWithAttributesChannel) { + ((DirectWithAttributesChannel) subscribableChannel).setAttribute("type", "output"); + this.messageChannelConfigurer.configureOutputChannel(subscribableChannel, name); + } + return subscribableChannel; } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java index db3802855..15762dfdf 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java @@ -50,6 +50,7 @@ import org.springframework.cloud.stream.binder.test.OutputDestination; import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration; import org.springframework.cloud.stream.binding.BindingsLifecycleController; import org.springframework.cloud.stream.binding.BindingsLifecycleController.State; +import org.springframework.cloud.stream.messaging.DirectWithAttributesChannel; import org.springframework.cloud.stream.messaging.Sink; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.Bean; @@ -61,6 +62,7 @@ import org.springframework.integration.support.MessageBuilder; import org.springframework.kafka.support.KafkaNull; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHeaders; +import org.springframework.messaging.SubscribableChannel; import org.springframework.messaging.support.GenericMessage; import org.springframework.scheduling.support.PeriodicTrigger; @@ -79,6 +81,23 @@ public class ImplicitFunctionBindingTests { System.clearProperty("spring.cloud.function.definition"); } + + @Test + public void testExplicitChannelConfiguration() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration(ExplicitChannelConfiguration.class)) + .web(WebApplicationType.NONE) + .run("--spring.jmx.enabled=false", "--spring.cloud.function.definition=echo")) { + + InputDestination input = context.getBean(InputDestination.class); + input.send(new GenericMessage("hello".getBytes()), "echo-in-0"); + + OutputDestination output = context.getBean(OutputDestination.class); + Message result = output.receive(1000, "echo-out-0"); + assertThat(result.getPayload()).isEqualTo("hello".getBytes()); + } + } + @SuppressWarnings({"rawtypes" }) @Test public void testDisableAutodetect() { @@ -1457,6 +1476,20 @@ public class ImplicitFunctionBindingTests { } } + @EnableAutoConfiguration + public static class ExplicitChannelConfiguration { + + @Bean("echo-in-0") + public SubscribableChannel myChannel() { + return new DirectWithAttributesChannel(); + } + + @Bean + public Function echo() { + return x -> x; + } + } + public static class Person { private String name; private int id;