From 5c510db012d46ce5d5f7ef4e9ee0d005dbd61460 Mon Sep 17 00:00:00 2001 From: Ilayaperumal Gopinathan Date: Wed, 20 Jul 2016 21:47:40 +0530 Subject: [PATCH] Aggregate application doesn't need binder in its classpath - BinderTypeRegistry will have empty binder types for aggregate applications - Check if `SharedChannelRegistry` bean is in the context, then proceed with no binder type This resolves #576 Support self contained aggregator app without binder dependency - Check if the aggregator application is self contained (with both source and sink applications exist) and if so, avoid having the check to have binder in classpath at BinderFactoryConfiguration - Given the aggregator parent context has binder configuration classes via `@EnableBinding`, enable all the auto configuration classes(excluding the binding configuration classes) as a parent for aggregator builder. This will have the aggregator parent to determine if the aggregator app is `selfContained` and set the parent context once all the app configurers are set - Add tests - Remove unused aggregator application methods --- .../aggregate/AggregateApplication.java | 80 +++++-------------- .../AggregateApplicationBuilder.java | 54 ++++++++----- .../config/BinderFactoryConfiguration.java | 12 ++- .../BinderFactoryConfigurationTests.java | 43 +++++++++- .../binder/InputOutputBindingOrderTest.java | 11 ++- .../partitioning/PartitionedConsumerTest.java | 12 ++- .../partitioning/PartitionedProducerTest.java | 9 ++- .../MockBinderRegistryConfiguration.java | 8 -- 8 files changed, 125 insertions(+), 104 deletions(-) diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/AggregateApplication.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/AggregateApplication.java index 4646f6d7e..d26cae6ae 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/AggregateApplication.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/AggregateApplication.java @@ -21,9 +21,6 @@ import java.util.Map.Entry; import org.springframework.boot.Banner.Mode; import org.springframework.boot.builder.SpringApplicationBuilder; -import org.springframework.cloud.stream.messaging.Processor; -import org.springframework.cloud.stream.messaging.Sink; -import org.springframework.cloud.stream.messaging.Source; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.integration.channel.DirectChannel; import org.springframework.messaging.SubscribableChannel; @@ -38,66 +35,38 @@ abstract class AggregateApplication { private static final String SPRING_CLOUD_STREAM_INTERNAL_PREFIX = "spring.cloud.stream.internal"; - public static final String CHANNEL_NAMESPACE_PROPERTY_NAME = - SPRING_CLOUD_STREAM_INTERNAL_PREFIX + ".channelNamespace"; + private static final String CHANNEL_NAMESPACE_PROPERTY_NAME = SPRING_CLOUD_STREAM_INTERNAL_PREFIX + ".channelNamespace"; + + private static final String SELF_CONTAINED_APP_PROPERTY_NAME = SPRING_CLOUD_STREAM_INTERNAL_PREFIX + ".selfContained"; public static final String INPUT_CHANNEL_NAME = "input"; public static final String OUTPUT_CHANNEL_NAME = "output"; - /** - * Supports the aggregation of {@link Source}, {@link Sink} and {@link Processor} - * apps by instantiating and binding them directly - * - * @param parentArgs arguments for the parent (prefixed with '--') - * @param apps a list app classes to be aggregated - * @param appArgs arguments for the apps (prefixed with '--") - * - * @return the resulting parent context for the aggregate - */ - static ConfigurableApplicationContext run(Class[] apps, String[] parentArgs, String[][] appArgs) { - ConfigurableApplicationContext parentContext = createParentContext(parentArgs != null ? parentArgs - : new String[0]); - runEmbedded(parentContext, apps, appArgs); - return parentContext; - } - - static ConfigurableApplicationContext run(Class... apps) { - return run(apps, null, null); - } - - /** - * Embeds a group of apps into an existing parent context - * - * @param parentContext the parent context - * @param apps a list of classes, representing root context definitions for apps - * @param args arguments for the apps - */ - static void runEmbedded(ConfigurableApplicationContext parentContext, - Class[] apps, String[][] args) { - SharedChannelRegistry bean = parentContext.getBean(SharedChannelRegistry.class); - prepareSharedChannelRegistry(bean, apps); - // create child contexts first - createChildContexts(parentContext, apps, args); - } - - static ConfigurableApplicationContext createParentContext(String[] args) { + static ConfigurableApplicationContext createParentContext(String[] args, boolean selfContained) { SpringApplicationBuilder aggregatorParentConfiguration = new SpringApplicationBuilder(); aggregatorParentConfiguration .sources(AggregatorParentConfiguration.class) .web(false) .headless(true) .properties("spring.jmx.default-domain=" - + AggregatorParentConfiguration.class.getName()); + + AggregatorParentConfiguration.class.getName(), + SELF_CONTAINED_APP_PROPERTY_NAME + "=" + selfContained); return aggregatorParentConfiguration.run(args); } - static void createChildContexts(ConfigurableApplicationContext parentContext, Class[] apps, String[][] args) { - for (int i = apps.length - 1; i >= 0; i--) { - String appClassName = apps[i].getName(); - embedApp(parentContext, getNamespace(appClassName, i), apps[i]).run(args != null ? args[i] : new - String[0]); - } + static ConfigurableApplicationContext createParentContext(ConfigurableApplicationContext parentContext, String[] args, + boolean selfContained) { + SpringApplicationBuilder aggregatorParentConfiguration = new SpringApplicationBuilder(); + aggregatorParentConfiguration + .sources(AggregatorParentConfiguration.class) + .web(false) + .headless(true) + .properties("spring.jmx.default-domain=" + + AggregatorParentConfiguration.class.getName(), + SELF_CONTAINED_APP_PROPERTY_NAME + "=" + selfContained) + .parent(parentContext); + return aggregatorParentConfiguration.run(args); } static String getNamespace(String appClassName, int index) { @@ -105,7 +74,7 @@ abstract class AggregateApplication { } protected static SpringApplicationBuilder embedApp( - ConfigurableApplicationContext applicationContext, String namespace, + ConfigurableApplicationContext parentContext, String namespace, Class app) { return new SpringApplicationBuilder(app) .web(false) @@ -114,16 +83,7 @@ abstract class AggregateApplication { .properties("spring.jmx.default-domain=" + app) .properties(CHANNEL_NAMESPACE_PROPERTY_NAME + "=" + namespace) .registerShutdownHook(false) - .parent(applicationContext); - } - - static void prepareSharedChannelRegistry(SharedChannelRegistry sharedChannelRegistry, Class[] apps) { - LinkedHashMap, String> appsToRegister = new LinkedHashMap<>(); - for (int i = apps.length - 1; i >= 0; i--) { - String appClassName = apps[i].getName(); - appsToRegister.put(apps[i], getNamespace(appClassName, i)); - } - prepareSharedChannelRegistry(sharedChannelRegistry, appsToRegister); + .parent(parentContext); } static void prepareSharedChannelRegistry(SharedChannelRegistry sharedChannelRegistry, 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 cc3d87fc9..73ccd1155 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 @@ -22,6 +22,7 @@ import java.util.LinkedHashMap; import java.util.List; import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; import org.springframework.boot.builder.SpringApplicationBuilder; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.util.Assert; @@ -46,36 +47,36 @@ public class AggregateApplicationBuilder { private AggregateApplicationBuilder applicationBuilder = this; - ConfigurableApplicationContext parentContext; - - public AggregateApplicationBuilder() { - this(SpringApplication.run(addAggregatorParentIfMissing(new Object[]{}), new String[]{})); - } + private ConfigurableApplicationContext parentContext; public AggregateApplicationBuilder(Object source, String... args) { this(new Object[]{source}, args); } public AggregateApplicationBuilder(Object[] sources, String[] args) { - this(SpringApplication.run(addAggregatorParentIfMissing(sources), args)); + this(SpringApplication.run(addParentConfiguration(sources), args)); } public AggregateApplicationBuilder(ConfigurableApplicationContext parentContext) { this.parentContext = parentContext; } - private static Object[] addAggregatorParentIfMissing(Object[] sources) { - Object[] aggregateParentSources; - if (!ObjectUtils.containsElement(sources, AggregatorParentConfiguration.class)) { - // add the AggregatorParentConfiguration first, so it can be overridden + /** + * Adding auto configuration classes to parent sources excluding the configuration + * classes related to binder/binding. + */ + private static Object[] addParentConfiguration(Object[] sources) { + Object[] parentSources; + if (!ObjectUtils.containsElement(sources, ParentConfiguration.class)) { + // add the ParentConfiguration first, so it can be overridden List sourceList = new ArrayList<>(Arrays.asList(sources)); - sourceList.add(0, AggregatorParentConfiguration.class); - aggregateParentSources = sourceList.toArray(new Object[sourceList.size()]); + sourceList.add(0, ParentConfiguration.class); + parentSources = sourceList.toArray(new Object[sourceList.size()]); } else { - aggregateParentSources = sources; + parentSources = sources; } - return aggregateParentSources; + return parentSources; } @@ -84,7 +85,7 @@ public class AggregateApplicationBuilder { } public AggregateApplicationBuilder parent(Object[] sources, String[] args) { - return parent(SpringApplication.run(addAggregatorParentIfMissing(sources), args)); + return parent(SpringApplication.run(sources, args)); } public AggregateApplicationBuilder parent(ConfigurableApplicationContext parentContext) { @@ -100,11 +101,6 @@ public class AggregateApplicationBuilder { } public ConfigurableApplicationContext run(String[] parentArgs) { - ConfigurableApplicationContext parentContext = this.parentContext != null - ? this.parentContext - : AggregateApplication.createParentContext(parentArgs); - SharedChannelRegistry sharedChannelRegistry = parentContext - .getBean(SharedChannelRegistry.class); List> apps = new ArrayList>(); if (this.sourceConfigurer != null) { apps.add(sourceConfigurer); @@ -127,12 +123,24 @@ public class AggregateApplicationBuilder { } appsToEmbed.put(appToEmbed, appConfigurer.namespace); } + if (this.parentContext == null) { + this.parentContext = AggregateApplication.createParentContext(parentArgs, areAppsSelfContained()); + } + // make sure to use the parent context that has aggregator parent configuration + else if (this.parentContext.getBeansOfType(SharedChannelRegistry.class).isEmpty()) { + this.parentContext = AggregateApplication.createParentContext(this.parentContext, parentArgs, areAppsSelfContained()); + } + SharedChannelRegistry sharedChannelRegistry = this.parentContext.getBean(SharedChannelRegistry.class); AggregateApplication.prepareSharedChannelRegistry(sharedChannelRegistry, appsToEmbed); for (int i = apps.size() - 1; i >= 0; i--) { AppConfigurer appConfigurer = apps.get(i); appConfigurer.embed(); } - return parentContext; + return this.parentContext; + } + + private boolean areAppsSelfContained() { + return (this.sourceConfigurer != null) && (this.sinkConfigurer != null); } private ChildContextBuilder childContext(Class app, @@ -281,4 +289,8 @@ public class AggregateApplicationBuilder { } + @EnableAutoConfiguration + public static class ParentConfiguration { + } + } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderFactoryConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderFactoryConfiguration.java index 8ef20f6f5..cf38d163e 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderFactoryConfiguration.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderFactoryConfiguration.java @@ -27,6 +27,7 @@ import java.util.Map; import java.util.Properties; import org.springframework.beans.factory.BeanCreationException; +import org.springframework.beans.factory.annotation.Value; import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; import org.springframework.cloud.stream.binder.BinderConfiguration; import org.springframework.cloud.stream.binder.BinderFactory; @@ -46,10 +47,18 @@ import org.springframework.util.StringUtils; /** * @author Marius Bogoevici + * @author Ilayaperumal Gopinathan */ @Configuration public class BinderFactoryConfiguration { + private static final String SPRING_CLOUD_STREAM_INTERNAL_PREFIX = "spring.cloud.stream.internal"; + + private static final String SELF_CONTAINED_APP_PROPERTY_NAME = SPRING_CLOUD_STREAM_INTERNAL_PREFIX + ".selfContained"; + + @Value("${" + SELF_CONTAINED_APP_PROPERTY_NAME + ":}") + private String selfContained; + @Bean @ConditionalOnMissingBean(BinderFactory.class) public BinderFactory binderFactory(BinderTypeRegistry binderTypeRegistry, @@ -100,7 +109,7 @@ public class BinderFactoryConfiguration { } try { Enumeration resources = classLoader.getResources("META-INF/spring.binders"); - if (resources == null || !resources.hasMoreElements()) { + if (!Boolean.valueOf(this.selfContained) && (resources == null || !resources.hasMoreElements())) { throw new BeanCreationException("Cannot create binder factory, no `META-INF/spring.binders` " + "resources found on the classpath"); } @@ -135,5 +144,4 @@ public class BinderFactoryConfiguration { } return parsedBinderConfigurations; } - } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderFactoryConfigurationTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderFactoryConfigurationTests.java index 55091870e..a841b5011 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderFactoryConfigurationTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderFactoryConfigurationTests.java @@ -25,6 +25,7 @@ import org.junit.Test; import org.springframework.beans.factory.BeanCreationException; import org.springframework.boot.autoconfigure.PropertyPlaceholderAutoConfiguration; import org.springframework.boot.builder.SpringApplicationBuilder; +import org.springframework.cloud.stream.aggregate.AggregatorParentConfiguration; import org.springframework.cloud.stream.annotation.EnableBinding; import org.springframework.cloud.stream.binder.stub1.StubBinder1; import org.springframework.cloud.stream.binder.stub1.StubBinder1Configuration; @@ -43,6 +44,7 @@ import static org.junit.Assert.fail; /** * @author Marius Bogoevici + * @author Ilayaperumal Gopinathan */ public class BinderFactoryConfigurationTests { @@ -58,6 +60,28 @@ public class BinderFactoryConfigurationTests { } } + @Test + public void loadBinderTypeRegistryWithNonSelfContainedAggregatorApp() throws Exception { + try { + createBinderTestContextWithSources( + new Class[]{SimpleApplication.class, AggregatorParentConfiguration.class}, new String[]{}, + "spring.cloud.stream.internal.selfContained=false"); + fail(); + } + catch (BeanCreationException e) { + assertThat(e.getMessage()).contains( + "Cannot create binder factory, no `META-INF/spring.binders` resources found on the classpath"); + } + + } + + @Test + public void loadBinderTypeRegistryWithSelfContainedAggregatorApp() throws Exception { + createBinderTestContextWithSources( + new Class[] { SimpleApplication.class, AggregatorParentConfiguration.class}, new String[] {}, + "spring.cloud.stream.internal.selfContained=true"); + } + @Test public void loadBinderTypeRegistryWithOneBinder() throws Exception { ConfigurableApplicationContext context = createBinderTestContext( @@ -208,7 +232,7 @@ public class BinderFactoryConfigurationTests { assertThat(defaultBinder).isSameAs(binder2); } - private static ConfigurableApplicationContext createBinderTestContext(String[] additionalClasspathDirectories, + private static ClassLoader createClassLoader(String[] additionalClasspathDirectories, String... properties) throws IOException { URL[] urls = ObjectUtils.isEmpty(additionalClasspathDirectories) ? new URL[0] : new URL[additionalClasspathDirectories.length]; @@ -217,7 +241,12 @@ public class BinderFactoryConfigurationTests { urls[i] = new URL(new ClassPathResource(additionalClasspathDirectories[i]).getURL().toString() + "/"); } } - ClassLoader classLoader = new URLClassLoader(urls, BinderFactoryConfigurationTests.class.getClassLoader()); + return new URLClassLoader(urls, BinderFactoryConfigurationTests.class.getClassLoader()); + } + + private static ConfigurableApplicationContext createBinderTestContext(String[] additionalClasspathDirectories, + String... properties) throws IOException { + ClassLoader classLoader = createClassLoader(additionalClasspathDirectories, properties); return new SpringApplicationBuilder(SimpleApplication.class) .resourceLoader(new DefaultResourceLoader(classLoader)) .properties(properties) @@ -225,6 +254,16 @@ public class BinderFactoryConfigurationTests { .run(); } + + private static ConfigurableApplicationContext createBinderTestContextWithSources(Class[] sources, String[] additionalClasspathDirectories, + String... properties) throws IOException { + ClassLoader classLoader = createClassLoader(additionalClasspathDirectories, properties); + return new SpringApplicationBuilder(sources) + .resourceLoader(new DefaultResourceLoader(classLoader)) + .properties(properties) + .web(false) + .run(); + } @Import({BinderFactoryConfiguration.class, PropertyPlaceholderAutoConfiguration.class}) @EnableBinding public static class SimpleApplication { diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/InputOutputBindingOrderTest.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/InputOutputBindingOrderTest.java index bf5fd5f4f..d460feaea 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/InputOutputBindingOrderTest.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/InputOutputBindingOrderTest.java @@ -23,6 +23,7 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; import org.springframework.cloud.stream.annotation.EnableBinding; +import org.springframework.cloud.stream.config.BinderFactoryConfiguration; import org.springframework.cloud.stream.messaging.Processor; import org.springframework.cloud.stream.utils.MockBinderRegistryConfiguration; import org.springframework.context.ConfigurableApplicationContext; @@ -38,6 +39,7 @@ import static org.mockito.Mockito.verifyNoMoreInteractions; /** * @author Marius Bogoevici + * @author Ilayaperumal Gopinathan */ public class InputOutputBindingOrderTest { @@ -58,7 +60,7 @@ public class InputOutputBindingOrderTest { @EnableBinding(Processor.class) @EnableAutoConfiguration - @Import(MockBinderRegistryConfiguration.class) + @Import({MockBinderRegistryConfiguration.class, BinderFactoryConfiguration.class}) public static class TestSource { @Bean @@ -71,7 +73,7 @@ public class InputOutputBindingOrderTest { @SuppressWarnings("rawtypes") @Autowired - private Binder binder; + private BinderFactory binderFactory; @Autowired private Processor processor; @@ -81,9 +83,10 @@ public class InputOutputBindingOrderTest { @Override @SuppressWarnings("unchecked") public synchronized void start() { - verify(this.binder).bindProducer(eq("output"), eq(this.processor.output()), Mockito.any()); + Binder binder = this.binderFactory.getBinder(null); + verify(binder).bindProducer(eq("output"), eq(this.processor.output()), Mockito.any()); // input was not bound yet - verifyNoMoreInteractions(this.binder); + verifyNoMoreInteractions(binder); this.running = true; } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/partitioning/PartitionedConsumerTest.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/partitioning/PartitionedConsumerTest.java index fb3c70694..451c68567 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/partitioning/PartitionedConsumerTest.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/partitioning/PartitionedConsumerTest.java @@ -28,7 +28,9 @@ import org.springframework.boot.test.context.SpringBootTest; import org.springframework.cloud.stream.annotation.Bindings; import org.springframework.cloud.stream.annotation.EnableBinding; import org.springframework.cloud.stream.binder.Binder; +import org.springframework.cloud.stream.binder.BinderFactory; import org.springframework.cloud.stream.binder.ConsumerProperties; +import org.springframework.cloud.stream.config.BinderFactoryConfiguration; import org.springframework.cloud.stream.messaging.Sink; import org.springframework.cloud.stream.utils.MockBinderRegistryConfiguration; import org.springframework.context.annotation.Import; @@ -43,6 +45,7 @@ import static org.mockito.Mockito.verifyNoMoreInteractions; /** * @author Marius Bogoevici + * @author Ilayaperumal Gopinathan */ @RunWith(SpringJUnit4ClassRunner.class) @SpringBootTest(classes = PartitionedConsumerTest.TestSink.class) @@ -50,7 +53,7 @@ public class PartitionedConsumerTest { @SuppressWarnings("rawtypes") @Autowired - private Binder binder; + private BinderFactory binderFactory; @Autowired @Bindings(TestSink.class) @@ -59,17 +62,18 @@ public class PartitionedConsumerTest { @Test @SuppressWarnings("unchecked") public void testBindingPartitionedConsumer() { + Binder binder = this.binderFactory.getBinder(null); ArgumentCaptor argumentCaptor = ArgumentCaptor.forClass(ConsumerProperties.class); - verify(this.binder).bindConsumer(eq("partIn"), anyString(), eq(this.testSink.input()), + verify(binder).bindConsumer(eq("partIn"), anyString(), eq(this.testSink.input()), argumentCaptor.capture()); Assert.assertThat(argumentCaptor.getValue().getInstanceIndex(), equalTo(0)); Assert.assertThat(argumentCaptor.getValue().getInstanceCount(), equalTo(2)); - verifyNoMoreInteractions(this.binder); + verifyNoMoreInteractions(binder); } @EnableBinding(Sink.class) @EnableAutoConfiguration - @Import(MockBinderRegistryConfiguration.class) + @Import({MockBinderRegistryConfiguration.class, BinderFactoryConfiguration.class}) @PropertySource("classpath:/org/springframework/cloud/stream/binder/partitioned-consumer-test.properties") public static class TestSink { diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/partitioning/PartitionedProducerTest.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/partitioning/PartitionedProducerTest.java index e3514d5e5..68c2e068a 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/partitioning/PartitionedProducerTest.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/partitioning/PartitionedProducerTest.java @@ -27,6 +27,7 @@ import org.springframework.boot.test.context.SpringBootTest; import org.springframework.cloud.stream.annotation.Bindings; import org.springframework.cloud.stream.annotation.EnableBinding; import org.springframework.cloud.stream.binder.Binder; +import org.springframework.cloud.stream.binder.BinderFactory; import org.springframework.cloud.stream.binder.ProducerProperties; import org.springframework.cloud.stream.messaging.Source; import org.springframework.cloud.stream.utils.MockBinderRegistryConfiguration; @@ -41,6 +42,7 @@ import static org.mockito.Mockito.verifyNoMoreInteractions; /** * @author Marius Bogoevici + * @author Ilayaperumal Gopinathan */ @RunWith(SpringJUnit4ClassRunner.class) @SpringBootTest(classes = PartitionedProducerTest.TestSource.class) @@ -48,7 +50,7 @@ public class PartitionedProducerTest { @SuppressWarnings("rawtypes") @Autowired - private Binder binder; + private BinderFactory binderFactory; @Autowired @Bindings(TestSource.class) @@ -57,12 +59,13 @@ public class PartitionedProducerTest { @Test @SuppressWarnings("unchecked") public void testBindingPartitionedProducer() { + Binder binder = this.binderFactory.getBinder(null); ArgumentCaptor argumentCaptor = ArgumentCaptor.forClass(ProducerProperties.class); - verify(this.binder).bindProducer(eq("partOut"), eq(this.testSource.output()), argumentCaptor.capture()); + verify(binder).bindProducer(eq("partOut"), eq(this.testSource.output()), argumentCaptor.capture()); Assert.assertThat(argumentCaptor.getValue().getPartitionCount(), equalTo(3)); Assert.assertThat(argumentCaptor.getValue().getPartitionKeyExpression().getExpressionString(), equalTo("payload")); - verifyNoMoreInteractions(this.binder); + verifyNoMoreInteractions(binder); } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/utils/MockBinderRegistryConfiguration.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/utils/MockBinderRegistryConfiguration.java index c41c391c4..3f9ba5850 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/utils/MockBinderRegistryConfiguration.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/utils/MockBinderRegistryConfiguration.java @@ -18,14 +18,11 @@ package org.springframework.cloud.stream.utils; import java.util.Collections; -import org.springframework.cloud.stream.binder.Binder; -import org.springframework.cloud.stream.binder.BinderFactory; import org.springframework.cloud.stream.binder.BinderType; import org.springframework.cloud.stream.binder.BinderTypeRegistry; import org.springframework.cloud.stream.binder.DefaultBinderTypeRegistry; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; -import org.springframework.messaging.MessageChannel; /** * A simple configuration that creates mock {@link org.springframework.cloud.stream.binder.Binder}s. @@ -39,9 +36,4 @@ public class MockBinderRegistryConfiguration { return new DefaultBinderTypeRegistry( Collections.singletonMap("mock", new BinderType("", new Class[] { MockBinderConfiguration.class }))); } - - @Bean - public Binder defaultBinder(BinderFactory binderFactory) { - return binderFactory.getBinder(null); - } }