SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Aggregate application doesn't need binder in its classpath

Browse Source 
- 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
This commit is contained in:
Ilayaperumal Gopinathan
2016-07-20 21:47:40 +05:30
committed by Soby Chacko
parent e462603dea
commit 5c510db012
8 changed files with 125 additions and 104 deletions
Download Patch File Download Diff File Expand all files Collapse all files

80
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/AggregateApplication.java
View File

@@ -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<Class<?>, 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,

54
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/AggregateApplicationBuilder.java
View File

@@ -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<Object> 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<AppConfigurer<?>> apps = new ArrayList<AppConfigurer<?>>();
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 {
}
}

12
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderFactoryConfiguration.java
View File

@@ -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<URL> 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;
}
}

43
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderFactoryConfigurationTests.java
View File

@@ -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 {

11
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/InputOutputBindingOrderTest.java
View File

@@ -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.<ProducerProperties>any());
Binder binder = this.binderFactory.getBinder(null);
verify(binder).bindProducer(eq("output"), eq(this.processor.output()), Mockito.<ProducerProperties>any());
// input was not bound yet
verifyNoMoreInteractions(this.binder);
verifyNoMoreInteractions(binder);
this.running = true;
}

12
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/partitioning/PartitionedConsumerTest.java
View File

@@ -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<ConsumerProperties> 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 {

9
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/partitioning/PartitionedProducerTest.java
View File

@@ -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<ProducerProperties> 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);
}

8
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/utils/MockBinderRegistryConfiguration.java
View File

@@ -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<MessageChannel> binderFactory) {
return binderFactory.getBinder(null);
}
}