diff --git a/spring-cloud-stream-test-support/src/test/java/org/springframework/cloud/stream/test/aggregate/main/AggregateWithMainTest.java b/spring-cloud-stream-test-support/src/test/java/org/springframework/cloud/stream/test/aggregate/main/AggregateWithMainTest.java index 870ac7eef..4eb43d46a 100644 --- a/spring-cloud-stream-test-support/src/test/java/org/springframework/cloud/stream/test/aggregate/main/AggregateWithMainTest.java +++ b/spring-cloud-stream-test-support/src/test/java/org/springframework/cloud/stream/test/aggregate/main/AggregateWithMainTest.java @@ -21,7 +21,8 @@ import java.util.concurrent.TimeUnit; import org.junit.Test; import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.boot.SpringBootConfiguration; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; import org.springframework.cloud.stream.aggregate.AggregateApplication; import org.springframework.cloud.stream.aggregate.AggregateApplicationBuilder; import org.springframework.cloud.stream.annotation.EnableBinding; @@ -37,6 +38,7 @@ import static org.assertj.core.api.Assertions.assertThat; /** * @author Marius Bogoevici + * @author Artem Bilan */ public class AggregateWithMainTest { @@ -59,13 +61,15 @@ public class AggregateWithMainTest { context.close(); } - @SpringBootApplication + @SpringBootConfiguration + @EnableAutoConfiguration public static class MainConfiguration { + } @Configuration @EnableBinding(Processor.class) - public static class UppercaseProcessor { + static class UppercaseProcessor { @Autowired Processor processor; @@ -74,15 +78,18 @@ public class AggregateWithMainTest { public String transform(String in) { return in.toUpperCase(); } + } @Configuration @EnableBinding(Processor.class) - public static class SuffixProcessor { + static class SuffixProcessor { @Transformer(inputChannel = Processor.INPUT, outputChannel = Processor.OUTPUT) public String transform(String in) { return in + " WORLD!"; } + } + } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/Input.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/Input.java index 35c4767b7..3c6026f8b 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/Input.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/Input.java @@ -1,5 +1,5 @@ /* - * Copyright 2015 the original author or authors. + * Copyright 2015-2017 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -40,6 +40,12 @@ import org.springframework.beans.factory.annotation.Qualifier; @Documented public @interface Input { + /** + * Specify the binding target name; + * used as a bean name for binding target + * and as a destination name by default. + * @return the binding target name + */ String value() default ""; } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/Output.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/Output.java index ba30f2432..d18a1f929 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/Output.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/Output.java @@ -1,5 +1,5 @@ /* - * Copyright 2015 the original author or authors. + * Copyright 2015-2017 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -30,6 +30,7 @@ import org.springframework.beans.factory.annotation.Qualifier; * * @author Dave Syer * @author Marius Bogoevici + * @author Artem Bilan */ @Qualifier @@ -40,6 +41,12 @@ import org.springframework.beans.factory.annotation.Qualifier; @Documented public @interface Output { + /** + * Specify the binding target name; + * used as a bean name for binding target + * and as a destination name by default. + * @return the binding target name + */ String value() default ""; } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingBeanDefinitionRegistryUtils.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingBeanDefinitionRegistryUtils.java index 626bc7454..f3c7ee47a 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingBeanDefinitionRegistryUtils.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingBeanDefinitionRegistryUtils.java @@ -1,5 +1,5 @@ /* - * Copyright 2015 the original author or authors. + * Copyright 2015-2017 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -20,6 +20,7 @@ import java.lang.annotation.Annotation; import java.lang.reflect.Method; import java.util.Map; +import org.springframework.beans.factory.BeanDefinitionStoreException; import org.springframework.beans.factory.support.AutowireCandidateQualifier; import org.springframework.beans.factory.support.BeanDefinitionRegistry; import org.springframework.beans.factory.support.RootBeanDefinition; @@ -36,6 +37,7 @@ import org.springframework.util.StringUtils; * * @author Marius Bogoevici * @author Dave Syer + * @author Artem Bilan */ public abstract class BindingBeanDefinitionRegistryUtils { @@ -57,6 +59,11 @@ public abstract class BindingBeanDefinitionRegistryUtils { String qualifierValue, String name, String bindingTargetInterfaceBeanName, String bindingTargetInterfaceMethodName, BeanDefinitionRegistry registry) { + if (registry.containsBeanDefinition(name)) { + throw new BeanDefinitionStoreException(bindingTargetInterfaceBeanName, name, + "bean definition with this name already exists - " + registry.getBeanDefinition(name)); + } + RootBeanDefinition rootBeanDefinition = new RootBeanDefinition(); rootBeanDefinition.setFactoryBeanName(bindingTargetInterfaceBeanName); rootBeanDefinition.setUniqueFactoryMethodName(bindingTargetInterfaceMethodName); diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingBeansRegistrar.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingBeansRegistrar.java index 222759ae1..8fab9ed09 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingBeansRegistrar.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingBeansRegistrar.java @@ -1,5 +1,5 @@ /* - * Copyright 2015 the original author or authors. + * Copyright 2015-2017 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -29,6 +29,7 @@ import org.springframework.util.ClassUtils; /** * @author Marius Bogoevici * @author Dave Syer + * @author Artem Bilan */ public class BindingBeansRegistrar implements ImportBeanDefinitionRegistrar { @@ -40,11 +41,13 @@ public class BindingBeansRegistrar implements ImportBeanDefinitionRegistrar { ClassUtils.resolveClassName(metadata.getClassName(), null), EnableBinding.class); for (Class type : collectClasses(attrs, metadata.getClassName())) { - BindingBeanDefinitionRegistryUtils.registerBindingTargetBeanDefinitions(type, - type.getName(), registry); - BindingBeanDefinitionRegistryUtils.registerBindingTargetsQualifiedBeanDefinitions( - ClassUtils.resolveClassName(metadata.getClassName(), null), type, - registry); + if (!registry.containsBeanDefinition(type.getName())) { + BindingBeanDefinitionRegistryUtils.registerBindingTargetBeanDefinitions(type, + type.getName(), registry); + BindingBeanDefinitionRegistryUtils.registerBindingTargetsQualifiedBeanDefinitions( + ClassUtils.resolveClassName(metadata.getClassName(), null), type, + registry); + } } } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/aggregation/AggregationTest.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/aggregation/AggregationTest.java index 932cbc5dd..73ae48389 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/aggregation/AggregationTest.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/aggregation/AggregationTest.java @@ -32,8 +32,8 @@ import org.springframework.boot.autoconfigure.web.EmbeddedServletContainerAutoCo import org.springframework.cloud.stream.aggregate.AggregateApplicationBuilder; import org.springframework.cloud.stream.aggregate.AggregateApplicationBuilder.SourceConfigurer; import org.springframework.cloud.stream.aggregate.SharedBindingTargetRegistry; -import org.springframework.cloud.stream.aggregate.SharedChannelRegistry; import org.springframework.cloud.stream.annotation.EnableBinding; +import org.springframework.cloud.stream.annotation.Output; import org.springframework.cloud.stream.binding.BindableProxyFactory; import org.springframework.cloud.stream.binding.BindingTargetFactory; import org.springframework.cloud.stream.messaging.Processor; @@ -51,6 +51,7 @@ import static org.junit.Assert.assertTrue; /** * @author Marius Bogoevici * @author Ilayaperumal Gopinathan + * @author Artem Bilan */ public class AggregationTest { @@ -84,13 +85,14 @@ public class AggregationTest { } @Test + @SuppressWarnings("deprecation") public void testModuleAggregationUsingSharedChannelRegistry() { // test backward compatibility aggregatedApplicationContext = new AggregateApplicationBuilder( MockBinderRegistryConfiguration.class, "--server.port=0") .from(TestSource.class).to(TestProcessor.class).run(); - SharedChannelRegistry sharedChannelRegistry = aggregatedApplicationContext - .getBean(SharedChannelRegistry.class); + org.springframework.cloud.stream.aggregate.SharedChannelRegistry sharedChannelRegistry = + aggregatedApplicationContext.getBean(org.springframework.cloud.stream.aggregate.SharedChannelRegistry.class); BindingTargetFactory channelFactory = aggregatedApplicationContext .getBean(BindingTargetFactory.class); assertThat(channelFactory).isNotNull(); @@ -99,6 +101,7 @@ public class AggregationTest { } @Test + @SuppressWarnings("unchecked") public void testParentArgsAndSources() { List argsToVerify = new ArrayList<>(); argsToVerify.add("--foo1=bar1"); @@ -113,7 +116,6 @@ public class AggregationTest { .namespace("foo").to(TestProcessor.class).namespace("bar") .run("--foo3=bar3", "--server.port=0"); DirectFieldAccessor aggregateApplicationBuilderAccessor = new DirectFieldAccessor(aggregateApplicationBuilder); - @SuppressWarnings("unchecked") final List parentArgs = (List) aggregateApplicationBuilderAccessor.getPropertyValue( "parentArgs"); assertThat(parentArgs).containsExactlyInAnyOrder(argsToVerify.toArray(new String[argsToVerify.size()])); @@ -125,8 +127,8 @@ public class AggregationTest { } @Test + @SuppressWarnings("unchecked") public void testParentArgsAndSourcesWithWebDisabled() { - List argsToVerify = new ArrayList<>(); AggregateApplicationBuilder aggregateApplicationBuilder = new AggregateApplicationBuilder( MockBinderRegistryConfiguration.class, "--foo1=bar1"); final ConfigurableApplicationContext context = aggregateApplicationBuilder @@ -155,7 +157,8 @@ public class AggregationTest { ((SourceConfigurer) aggregateApplicationBuilderAccessor.getPropertyValue("sourceConfigurer")) .getArgs(), new String[] { "--foo1=bar1" })); - final List processorConfigurers = (List) aggregateApplicationBuilderAccessor + final List processorConfigurers = + (List) aggregateApplicationBuilderAccessor .getPropertyValue("processorConfigurers"); for (AggregateApplicationBuilder.ProcessorConfigurer processorConfigurer : processorConfigurers) { if (processorConfigurer.getNamespace().equals("b")) { @@ -185,7 +188,8 @@ public class AggregationTest { ((SourceConfigurer) aggregateApplicationBuilderAccessor.getPropertyValue("sourceConfigurer")) .getArgs(), new String[] { "--fooValue=bara" })); - final List processorConfigurers = (List) aggregateApplicationBuilderAccessor + final List processorConfigurers = + (List) aggregateApplicationBuilderAccessor .getPropertyValue("processorConfigurers"); for (AggregateApplicationBuilder.ProcessorConfigurer processorConfigurer : processorConfigurers) { if (processorConfigurer.getNamespace().equals("b")) { @@ -215,7 +219,8 @@ public class AggregationTest { ((SourceConfigurer) aggregateApplicationBuilderAccessor.getPropertyValue("sourceConfigurer")) .getArgs(), new String[] { "--fooValue=bara" })); - final List processorConfigurers = (List) aggregateApplicationBuilderAccessor + final List processorConfigurers = + (List) aggregateApplicationBuilderAccessor .getPropertyValue("processorConfigurers"); for (AggregateApplicationBuilder.ProcessorConfigurer processorConfigurer : processorConfigurers) { if (processorConfigurer.getNamespace().equals("b")) { @@ -247,7 +252,8 @@ public class AggregationTest { ((SourceConfigurer) aggregateApplicationBuilderAccessor.getPropertyValue("sourceConfigurer")) .getArgs(), new String[] { "--fooValue=bara" })); - final List processorConfigurers = (List) aggregateApplicationBuilderAccessor + final List processorConfigurers = + (List) aggregateApplicationBuilderAccessor .getPropertyValue("processorConfigurers"); for (AggregateApplicationBuilder.ProcessorConfigurer processorConfigurer : processorConfigurers) { if (processorConfigurer.getNamespace().equals("b")) { @@ -279,9 +285,8 @@ public class AggregationTest { ((SourceConfigurer) aggregateApplicationBuilderAccessor.getPropertyValue("sourceConfigurer")) .getArgs(), new String[] { "--foo-value=sysbara" })); - final List processorConfigurers = (List) aggregateApplicationBuilderAccessor - .getPropertyValue("processorConfigurers"); - for (AggregateApplicationBuilder.ProcessorConfigurer processorConfigurer : ((List) aggregateApplicationBuilderAccessor + for (AggregateApplicationBuilder.ProcessorConfigurer processorConfigurer : + ((List) aggregateApplicationBuilderAccessor .getPropertyValue( "processorConfigurers"))) { if (processorConfigurer.getNamespace().equals("b")) { @@ -329,13 +334,14 @@ public class AggregationTest { } @Test + @SuppressWarnings("deprecation") public void testNamespaces() { aggregatedApplicationContext = new AggregateApplicationBuilder( MockBinderRegistryConfiguration.class, "--server.port=0") .from(TestSource.class).namespace("foo").to(TestProcessor.class) .namespace("bar").run(); - SharedChannelRegistry sharedChannelRegistry = aggregatedApplicationContext - .getBean(SharedChannelRegistry.class); + org.springframework.cloud.stream.aggregate.SharedChannelRegistry sharedChannelRegistry = + aggregatedApplicationContext.getBean(org.springframework.cloud.stream.aggregate.SharedChannelRegistry.class); BindingTargetFactory channelFactory = aggregatedApplicationContext .getBean(BindingTargetFactory.class); Object fooOutput = sharedChannelRegistry.get("foo.output"); @@ -353,16 +359,22 @@ public class AggregationTest { public void testBindableProxyFactoryCaching() { AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext( MockBinderRegistryConfiguration.class, - TestSource.class, TestProcessor.class); + TestSource2.class, TestProcessor.class); Map factories = context.getBeansOfType(BindableProxyFactory.class); assertThat(factories).hasSize(2); Map sources = context.getBeansOfType(Source.class); - assertThat(sources).hasSize(2); + assertThat(sources).hasSize(1); for (Source source : sources.values()) { source.output(); } + Map fooSources = context.getBeansOfType(FooSource.class); + assertThat(fooSources).hasSize(1); + for (FooSource source : fooSources.values()) { + source.output(); + } + Map processors = context.getBeansOfType(Processor.class); assertThat(processors).hasSize(1); for (Processor processor : processors.values()) { @@ -377,6 +389,9 @@ public class AggregationTest { if (factory.getObjectType() == Source.class) { assertThat(targetCache).hasSize(1); } + if (factory.getObjectType() == FooSource.class) { + assertThat(targetCache).hasSize(1); + } else if (factory.getObjectType() == Processor.class) { assertThat(targetCache).hasSize(2); } @@ -399,6 +414,20 @@ public class AggregationTest { } + public interface FooSource { + + @Output("fooOutput") + MessageChannel output(); + + } + + @EnableBinding(FooSource.class) + @EnableAutoConfiguration + public static class TestSource2 { + + } + + @Configuration public static class DummyConfig { diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/InvalidBindingConfigurationTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/InvalidBindingConfigurationTests.java new file mode 100644 index 000000000..b21e441b2 --- /dev/null +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/InvalidBindingConfigurationTests.java @@ -0,0 +1,79 @@ +/* + * Copyright 2017 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.binding; + +import org.assertj.core.api.ThrowableAssert; +import org.junit.Test; + +import org.springframework.beans.factory.BeanDefinitionStoreException; +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.cloud.stream.annotation.EnableBinding; +import org.springframework.cloud.stream.annotation.Input; +import org.springframework.cloud.stream.annotation.Output; +import org.springframework.cloud.stream.utils.MockBinderRegistryConfiguration; +import org.springframework.context.annotation.Import; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.SubscribableChannel; + +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** + * @author Artem Bilan + * + * @since 1.3 + */ +public class InvalidBindingConfigurationTests { + + @Test + public void testDuplicateBeanByBindingConfig() { + assertThatThrownBy( + new ThrowableAssert.ThrowingCallable() { + + @Override + public void call() throws Throwable { + SpringApplication.run(TestBindingConfig.class); + } + + }) + .isInstanceOf(BeanDefinitionStoreException.class) + .hasMessageContaining("bean definition with this name already exists") + .hasMessageContaining(TestInvalidBinding.NAME) + .hasNoCause(); + } + + @EnableBinding(TestInvalidBinding.class) + @EnableAutoConfiguration + @Import(MockBinderRegistryConfiguration.class) + public static class TestBindingConfig { + + } + + + public interface TestInvalidBinding { + + String NAME = "testName"; + + @Input(NAME) + SubscribableChannel in(); + + @Output(NAME) + MessageChannel out(); + + } + +}