diff --git a/spring-cloud-stream-codec/src/main/java/org/springframework/cloud/stream/config/codec/kryo/KryoCodecAutoConfiguration.java b/spring-cloud-stream-codec/src/main/java/org/springframework/cloud/stream/config/codec/kryo/KryoCodecAutoConfiguration.java index 4b3e72a9e..8985f06eb 100644 --- a/spring-cloud-stream-codec/src/main/java/org/springframework/cloud/stream/config/codec/kryo/KryoCodecAutoConfiguration.java +++ b/spring-cloud-stream-codec/src/main/java/org/springframework/cloud/stream/config/codec/kryo/KryoCodecAutoConfiguration.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. @@ -51,6 +51,7 @@ public class KryoCodecAutoConfiguration { KryoCodecProperties kryoCodecProperties; @Bean + @ConditionalOnMissingBean(PojoCodec.class) public PojoCodec codec() { Map kryoRegistrarMap = applicationContext.getBeansOfType(KryoRegistrar .class); @@ -58,6 +59,7 @@ public class KryoCodecAutoConfiguration { } @Bean + @ConditionalOnMissingBean(KryoRegistrar.class) public KryoRegistrar fileRegistrar() { return new FileKryoRegistrar(); } diff --git a/spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/ReactiveSupportAutoConfiguration.java b/spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/ReactiveSupportAutoConfiguration.java index 22194595c..f13b504e9 100644 --- a/spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/ReactiveSupportAutoConfiguration.java +++ b/spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/ReactiveSupportAutoConfiguration.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-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. @@ -16,11 +16,9 @@ package org.springframework.cloud.stream.reactive; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; - import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; +import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; import org.springframework.cloud.stream.binding.BindingService; import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory; import org.springframework.context.annotation.Bean; @@ -33,9 +31,8 @@ import org.springframework.context.annotation.Configuration; @ConditionalOnBean(BindingService.class) public class ReactiveSupportAutoConfiguration { - private static Log log = LogFactory.getLog(ReactiveSupportAutoConfiguration.class); - @Bean + @ConditionalOnMissingBean(MessageChannelToInputFluxParameterAdapter.class) public MessageChannelToInputFluxParameterAdapter messageChannelToInputFluxArgumentAdapter( CompositeMessageConverterFactory compositeMessageConverterFactory) { return new MessageChannelToInputFluxParameterAdapter( @@ -43,11 +40,13 @@ public class ReactiveSupportAutoConfiguration { } @Bean + @ConditionalOnMissingBean(MessageChannelToFluxSenderParameterAdapter.class) public MessageChannelToFluxSenderParameterAdapter messageChannelToFluxSenderArgumentAdapter() { return new MessageChannelToFluxSenderParameterAdapter(); } @Bean + @ConditionalOnMissingBean(FluxToMessageChannelResultAdapter.class) public FluxToMessageChannelResultAdapter fluxToMessageChannelResultAdapter() { return new FluxToMessageChannelResultAdapter(); } @@ -57,18 +56,21 @@ public class ReactiveSupportAutoConfiguration { public static class RxJava1SupportConfiguration { @Bean + @ConditionalOnMissingBean(MessageChannelToInputObservableParameterAdapter.class) public MessageChannelToInputObservableParameterAdapter messageChannelToInputObservableArgumentAdapter( MessageChannelToInputFluxParameterAdapter messageChannelToFluxArgumentAdapter) { return new MessageChannelToInputObservableParameterAdapter(messageChannelToFluxArgumentAdapter); } @Bean + @ConditionalOnMissingBean(MessageChannelToObservableSenderParameterAdapter.class) public MessageChannelToObservableSenderParameterAdapter messageChannelToObservableSenderArgumentAdapter( MessageChannelToFluxSenderParameterAdapter messageChannelToFluxSenderArgumentAdapter) { return new MessageChannelToObservableSenderParameterAdapter(messageChannelToFluxSenderArgumentAdapter); } @Bean + @ConditionalOnMissingBean(ObservableToMessageChannelResultAdapter.class) public ObservableToMessageChannelResultAdapter observableToMessageChannelResultAdapter( FluxToMessageChannelResultAdapter fluxToMessageChannelResultAdapter) { diff --git a/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroMessageConverterAutoConfiguration.java b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroMessageConverterAutoConfiguration.java index d4aeed817..753ec7371 100644 --- a/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroMessageConverterAutoConfiguration.java +++ b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroMessageConverterAutoConfiguration.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-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. @@ -19,6 +19,7 @@ package org.springframework.cloud.stream.schema.avro; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; +import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.cloud.stream.binder.StringConvertingContentTypeResolver; @@ -42,6 +43,7 @@ public class AvroMessageConverterAutoConfiguration { private AvroMessageConverterProperties avroMessageConverterProperties; @Bean + @ConditionalOnMissingBean(AvroSchemaRegistryClientMessageConverter.class) public AvroSchemaRegistryClientMessageConverter avroSchemaMessageConverter( SchemaRegistryClient schemaRegistryClient) { AvroSchemaRegistryClientMessageConverter diff --git a/spring-cloud-stream-test-support/src/test/java/org/springframework/cloud/stream/test/aggregate/AggregateTestWithMain.java b/spring-cloud-stream-test-support/src/test/java/org/springframework/cloud/stream/test/aggregate/AggregateTestWithMain.java index fa3a66fc0..d3b4867b5 100644 --- a/spring-cloud-stream-test-support/src/test/java/org/springframework/cloud/stream/test/aggregate/AggregateTestWithMain.java +++ b/spring-cloud-stream-test-support/src/test/java/org/springframework/cloud/stream/test/aggregate/AggregateTestWithMain.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-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. @@ -25,6 +25,7 @@ import org.springframework.cloud.stream.aggregate.AggregateApplicationBuilder; import org.springframework.cloud.stream.annotation.EnableBinding; import org.springframework.cloud.stream.messaging.Processor; import org.springframework.cloud.stream.test.binder.MessageCollector; +import org.springframework.cloud.stream.test.binder.TestSupportBinderAutoConfiguration; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.Configuration; import org.springframework.integration.annotation.Transformer; @@ -41,7 +42,7 @@ public class AggregateTestWithMain { @Test public void testAggregateApplication() throws InterruptedException { // emulate a main method - ConfigurableApplicationContext context = new AggregateApplicationBuilder().from(UppercaseProcessor.class) + ConfigurableApplicationContext context = new AggregateApplicationBuilder(TestSupportBinderAutoConfiguration.class).from(UppercaseProcessor.class) .namespace("upper").to(SuffixProcessor.class).namespace("suffix").run(); AggregateApplication aggregateAccessor = context.getBean(AggregateApplication.class); diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/AggregateApplicationBuilder.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/AggregateApplicationBuilder.java index 87318115f..90d4eb5d1 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/AggregateApplicationBuilder.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/AggregateApplicationBuilder.java @@ -29,16 +29,19 @@ import java.util.Set; import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanFactoryUtils; import org.springframework.beans.factory.SmartInitializingSingleton; +import org.springframework.boot.actuate.autoconfigure.EndpointAutoConfiguration; import org.springframework.boot.actuate.endpoint.MetricReaderPublicMetrics; import org.springframework.boot.actuate.endpoint.MetricsEndpoint; -import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.autoconfigure.ImportAutoConfiguration; import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; +import org.springframework.boot.autoconfigure.web.EmbeddedServletContainerAutoConfiguration; import org.springframework.boot.bind.PropertySourcesPropertyValues; import org.springframework.boot.bind.RelaxedDataBinder; import org.springframework.boot.bind.RelaxedNames; import org.springframework.boot.builder.SpringApplicationBuilder; import org.springframework.cloud.stream.annotation.EnableBinding; import org.springframework.cloud.stream.binding.BindableProxyFactory; +import org.springframework.cloud.stream.config.ChannelBindingAutoConfiguration; import org.springframework.context.ApplicationContext; import org.springframework.context.ApplicationContextAware; import org.springframework.context.ConfigurableApplicationContext; @@ -77,11 +80,11 @@ public class AggregateApplicationBuilder implements AggregateApplication, Applic private boolean webEnvironment = true; public AggregateApplicationBuilder(String... args) { - this(new Object[] { ParentConfiguration.class }, args); + this(new Object[]{ParentConfiguration.class}, args); } public AggregateApplicationBuilder(Object source, String... args) { - this(new Object[] { source }, args); + this(new Object[]{source}, args); } public AggregateApplicationBuilder(Object[] sources, String[] args) { @@ -101,7 +104,7 @@ public class AggregateApplicationBuilder implements AggregateApplication, Applic } public AggregateApplicationBuilder parent(Object source, String... args) { - return parent(new Object[] { source }, args); + return parent(new Object[]{source}, args); } public AggregateApplicationBuilder parent(Object[] sources, String[] args) { @@ -151,7 +154,8 @@ public class AggregateApplicationBuilder implements AggregateApplication, Applic } try { return bindableType.cast(parentContext.getBean(namespace + "." + bindableType.getName())); - } catch (BeansException e) { + } + catch (BeansException e) { throw new IllegalStateException("Binding not found for '" + bindableType.getName() + "' into namespace " + namespace); } @@ -190,6 +194,9 @@ public class AggregateApplicationBuilder implements AggregateApplication, Applic appConfigurers.put(appConfigurer, appConfigurer.namespace); } if (this.parentContext == null) { + if (Boolean.TRUE.equals(this.webEnvironment)) { + this.addParentSources(new Object[]{EmbeddedServletContainerAutoConfiguration.class}); + } this.parentContext = AggregateApplicationUtils.createParentContext(this.parentSources.toArray(new Object[0]), this.parentArgs.toArray(new String[0]), selfContained(), this.webEnvironment, this.headless); } @@ -369,7 +376,7 @@ public class AggregateApplicationBuilder implements AggregateApplication, Applic void embed() { final ConfigurableApplicationContext childContext = childContext(this.app, AggregateApplicationBuilder.this.parentContext, this.namespace).args(this.args).config(this.names) - .profiles(this.profiles).run(); + .profiles(this.profiles).run(); // Register bindable proxies as beans so they can be queried for later Map bindableProxies = BeanFactoryUtils .beansOfTypeIncludingAncestors(childContext.getBeanFactory(), BindableProxyFactory.class); @@ -455,7 +462,7 @@ public class AggregateApplicationBuilder implements AggregateApplication, Applic } - @EnableAutoConfiguration + @ImportAutoConfiguration({ChannelBindingAutoConfiguration.class, EndpointAutoConfiguration.class}) @EnableBinding public static class ParentConfiguration { @Bean diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java index 133daa316..7ca343337 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2016 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. @@ -101,18 +101,21 @@ public class BindingServiceConfiguration { } @Bean + @ConditionalOnMissingBean(MessageConverterConfigurer.class) public MessageConverterConfigurer messageConverterConfigurer(BindingServiceProperties bindingServiceProperties, CompositeMessageConverterFactory compositeMessageConverterFactory) { return new MessageConverterConfigurer(bindingServiceProperties, compositeMessageConverterFactory); } @Bean + @ConditionalOnMissingBean(SubscribableChannelBindingTargetFactory.class) public SubscribableChannelBindingTargetFactory channelFactory( CompositeMessageChannelConfigurer compositeMessageChannelConfigurer) { return new SubscribableChannelBindingTargetFactory(compositeMessageChannelConfigurer); } @Bean + @ConditionalOnMissingBean(CompositeMessageChannelConfigurer.class) public CompositeMessageChannelConfigurer compositeMessageChannelConfigurer( MessageConverterConfigurer messageConverterConfigurer) { List configurerList = new ArrayList<>(); @@ -122,23 +125,27 @@ public class BindingServiceConfiguration { @Bean @DependsOn("bindingService") + @ConditionalOnMissingBean(OutputBindingLifecycle.class) public OutputBindingLifecycle outputBindingLifecycle() { return new OutputBindingLifecycle(); } @Bean @DependsOn("bindingService") + @ConditionalOnMissingBean(InputBindingLifecycle.class) public InputBindingLifecycle inputBindingLifecycle() { return new InputBindingLifecycle(); } @Bean @DependsOn("bindingService") + @ConditionalOnMissingBean(ContextStartAfterRefreshListener.class) public ContextStartAfterRefreshListener contextStartAfterRefreshListener() { return new ContextStartAfterRefreshListener(); } @Bean + @ConditionalOnMissingBean(BinderAwareChannelResolver.class) public BinderAwareChannelResolver binderAwareChannelResolver(BindingService bindingService, AbstractBindingTargetFactory bindingTargetFactory, DynamicDestinationsBindable dynamicDestinationsBindable) { @@ -147,17 +154,20 @@ public class BindingServiceConfiguration { @Bean @ConditionalOnProperty("spring.cloud.stream.bindings." + ERROR_CHANNEL_NAME + ".destination") + @ConditionalOnMissingBean(SingleBindingTargetBindable.class) public SingleBindingTargetBindable errorChannelBindable( @Qualifier(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME) PublishSubscribeChannel errorChannel) { return new SingleBindingTargetBindable(ERROR_CHANNEL_NAME, errorChannel); } @Bean + @ConditionalOnMissingBean(DynamicDestinationsBindable.class) public DynamicDestinationsBindable dynamicDestinationsBindable() { return new DynamicDestinationsBindable(); } @Bean + @ConditionalOnMissingBean(CompositeMessageConverterFactory.class) public CompositeMessageConverterFactory compositeMessageConverterFactory() { List messageConverters = new ArrayList<>(); if (!CollectionUtils.isEmpty(this.customMessageConverters)) { @@ -167,6 +177,7 @@ public class BindingServiceConfiguration { } @Bean + @ConditionalOnMissingBean(MessageHandlerMethodFactory.class) public static MessageHandlerMethodFactory messageHandlerMethodFactory( CompositeMessageConverterFactory compositeMessageConverterFactory) { DefaultMessageHandlerMethodFactory messageHandlerMethodFactory = new DefaultMessageHandlerMethodFactory(); @@ -176,6 +187,7 @@ public class BindingServiceConfiguration { } @Bean + @ConditionalOnMissingBean(StreamListenerAnnotationBeanPostProcessor.class) public static StreamListenerAnnotationBeanPostProcessor bindToAnnotationBeanPostProcessor( @Lazy BinderAwareChannelResolver binderAwareChannelResolver, @Lazy MessageHandlerMethodFactory messageHandlerMethodFactory) { @@ -196,6 +208,7 @@ public class BindingServiceConfiguration { private BinderAwareChannelResolver binderAwareChannelResolver; @Bean + @ConditionalOnMissingBean(BinderAwareRouterBeanPostProcessor.class) public BinderAwareRouterBeanPostProcessor binderAwareRouterBeanPostProcessor( final ConfigurableListableBeanFactory beanFactory) { // IMPORTANT: Lazy delegate to avoid instantiating all of the above early 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 f310ea322..f0705a6c2 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 @@ -28,6 +28,7 @@ import org.junit.Test; import org.springframework.beans.DirectFieldAccessor; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.autoconfigure.web.EmbeddedServletContainerAutoConfiguration; import org.springframework.cloud.stream.aggregate.AggregateApplicationBuilder; import org.springframework.cloud.stream.aggregate.AggregateApplicationBuilder.SourceConfigurer; import org.springframework.cloud.stream.aggregate.SharedBindingTargetRegistry; @@ -69,7 +70,7 @@ public class AggregationTest { @Test public void aggregation() { aggregatedApplicationContext = new AggregateApplicationBuilder( - MockBinderRegistryConfiguration.class, "--server.port=0") + MockBinderRegistryConfiguration.class, "--server.port=0", "--debug=true") .from(TestSource.class) .to(TestProcessor.class) .run(); @@ -118,6 +119,23 @@ public class AggregationTest { "parentArgs"); assertThat(parentArgs).containsExactlyInAnyOrder(argsToVerify.toArray(new String[argsToVerify.size()])); List sources = (List) aggregateApplicationBuilderAccessor.getPropertyValue("parentSources"); + assertThat(sources).containsExactlyInAnyOrder(AggregateApplicationBuilder.ParentConfiguration.class, + MockBinderRegistryConfiguration.class, DummyConfig.class, EmbeddedServletContainerAutoConfiguration.class); + context.close(); + } + + @Test + public void testParentArgsAndSourcesWithWebDisabled() { + List argsToVerify = new ArrayList<>(); + AggregateApplicationBuilder aggregateApplicationBuilder = + new AggregateApplicationBuilder(MockBinderRegistryConfiguration.class, "--foo1=bar1"); + final ConfigurableApplicationContext context = + aggregateApplicationBuilder.parent(DummyConfig.class, "--foo2=bar2").web(false) + .from(TestSource.class) + .namespace("foo").to(TestProcessor.class).namespace("bar") + .run("--server.port=0"); + DirectFieldAccessor aggregateApplicationBuilderAccessor = new DirectFieldAccessor(aggregateApplicationBuilder); + List sources = (List) aggregateApplicationBuilderAccessor.getPropertyValue("parentSources"); assertThat(sources).containsExactlyInAnyOrder(AggregateApplicationBuilder.ParentConfiguration.class, MockBinderRegistryConfiguration.class, DummyConfig.class); context.close();