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

Improve application aggregation and testability

Browse Source 
Fixes #723

- Add support for registering an `AggregateApplication` bean for accessing the
  components from underlying subcontexts
- Testing support and samples

Signed-off-by: Marius Bogoevici <mbogoevici@pivotal.io>

Fix Javadoc
This commit is contained in:
Marius Bogoevici
2016-11-30 11:33:23 -05:00
committed by Ilayaperumal Gopinathan
parent 8fa41a361f
commit 6c861135a3
8 changed files with 342 additions and 108 deletions
Download Patch File Download Diff File Expand all files Collapse all files

78
spring-cloud-stream-test-support/src/test/java/org/springframework/cloud/stream/test/aggregate/AggregateTestWithBean.java Normal file
View File

@@ -0,0 +1,78 @@
package org.springframework.cloud.stream.test.aggregate;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.cloud.stream.aggregate.AggregateApplication;
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.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.Transformer;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Marius Bogoevici
*/
@RunWith(SpringJUnit4ClassRunner.class)
@SpringBootTest(classes = AggregateTestWithBean.ChainedProcessors.class, properties = {"server.port=-1"})
public class AggregateTestWithBean {
@Autowired
public MessageCollector messageCollector;
@Autowired
public AggregateApplication aggregateApplication;
@Test
public void testAggregateApplication() throws InterruptedException {
Processor uppercaseProcessor = aggregateApplication.getBinding(Processor.class, "upper");
Processor suffixProcessor = aggregateApplication.getBinding(Processor.class, "suffix");
uppercaseProcessor.input().send(MessageBuilder.withPayload("Hello").build());
Message<?> receivedMessage = messageCollector.forChannel(suffixProcessor.output()).poll(1, TimeUnit.SECONDS);
assertThat(receivedMessage).isNotNull();
assertThat(receivedMessage.getPayload()).isEqualTo("HELLO WORLD!");
}
@SpringBootApplication
@EnableBinding
public static class ChainedProcessors {
@Bean
public AggregateApplication aggregateApplication() {
return new AggregateApplicationBuilder().from(UppercaseProcessor.class)
.namespace("upper").to(SuffixProcessor.class).namespace("suffix").build();
}
}
@Configuration
@EnableBinding(Processor.class)
public static class UppercaseProcessor {
@Transformer(inputChannel = Processor.INPUT, outputChannel = Processor.OUTPUT)
public String transform(String in) {
return in.toUpperCase();
}
}
@Configuration
@EnableBinding(Processor.class)
public static class SuffixProcessor {
@Transformer(inputChannel = Processor.INPUT, outputChannel = Processor.OUTPUT)
public String transform(String in) {
return in + " WORLD!";
}
}
}

61
spring-cloud-stream-test-support/src/test/java/org/springframework/cloud/stream/test/aggregate/AggregateTestWithMain.java Normal file
View File

@@ -0,0 +1,61 @@
package org.springframework.cloud.stream.test.aggregate;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import org.springframework.cloud.stream.aggregate.AggregateApplication;
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.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.Transformer;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Marius Bogoevici
*/
public class AggregateTestWithMain {
@Test
public void testAggregateApplication() throws InterruptedException {
// emulate a main method
ConfigurableApplicationContext context = new AggregateApplicationBuilder().from(UppercaseProcessor.class)
.namespace("upper").to(SuffixProcessor.class).namespace("suffix").run();
AggregateApplication aggregateAccessor = context.getBean(AggregateApplication.class);
MessageCollector messageCollector = context.getBean(MessageCollector.class);
Processor uppercaseProcessor = aggregateAccessor.getBinding(Processor.class, "upper");
Processor suffixProcessor = aggregateAccessor.getBinding(Processor.class, "suffix");
uppercaseProcessor.input().send(MessageBuilder.withPayload("Hello").build());
Message<?> receivedMessage = messageCollector.forChannel(suffixProcessor.output()).poll(1, TimeUnit.SECONDS);
assertThat(receivedMessage).isNotNull();
assertThat(receivedMessage.getPayload()).isEqualTo("HELLO WORLD!");
context.close();
}
@Configuration
@EnableBinding(Processor.class)
public static class UppercaseProcessor {
@Transformer(inputChannel = Processor.INPUT, outputChannel = Processor.OUTPUT)
public String transform(String in) {
return in.toUpperCase();
}
}
@Configuration
@EnableBinding(Processor.class)
public static class SuffixProcessor {
@Transformer(inputChannel = Processor.INPUT, outputChannel = Processor.OUTPUT)
public String transform(String in) {
return in + " WORLD!";
}
}
}

2
spring-cloud-stream-test-support/src/test/java/org/springframework/cloud/stream/test/ExampleTest.java → spring-cloud-stream-test-support/src/test/java/org/springframework/cloud/stream/test/example/ExampleTest.java
View File

@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.cloud.stream.test;
package org.springframework.cloud.stream.test.example;
import org.junit.Test;
import org.junit.runner.RunWith;

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

@@ -1,98 +1,23 @@
/*
* Copyright 2015-2016 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.aggregate;
import java.util.LinkedHashMap;
import java.util.Map.Entry;
import org.springframework.boot.Banner.Mode;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.messaging.SubscribableChannel;
/**
* Class that is responsible for embedding apps using shared channel registry.
* Handle to an aggregate application, providing access to the underlying
* components of the aggregate (e.g. bindable instances).
*
* @author Marius Bogoevici
* @author Ilayaperumal Gopinathan
* @author Venil Noronha
*/
abstract class AggregateApplication {
private static final String SPRING_CLOUD_STREAM_INTERNAL_PREFIX = "spring.cloud.stream.internal";
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";
static ConfigurableApplicationContext createParentContext(Object[] sources, String[] args, final
boolean selfContained, boolean webEnvironment,
boolean headless) {
SpringApplicationBuilder aggregatorParentConfiguration = new SpringApplicationBuilder();
aggregatorParentConfiguration
.sources(sources)
.web(webEnvironment)
.headless(headless)
.properties("spring.jmx.default-domain="
+ AggregateApplicationBuilder.ParentConfiguration.class.getName(),
SELF_CONTAINED_APP_PROPERTY_NAME + "=" + selfContained);
return aggregatorParentConfiguration.run(args);
}
static String getDefaultNamespace(String appClassName, int index) {
return appClassName + "_" + index;
}
protected static SpringApplicationBuilder embedApp(
ConfigurableApplicationContext parentContext, String namespace,
Class<?> app) {
return new SpringApplicationBuilder(app)
.web(false)
.main(app)
.bannerMode(Mode.OFF)
.properties("spring.jmx.default-domain=" + namespace)
.properties(CHANNEL_NAMESPACE_PROPERTY_NAME + "=" + namespace)
.registerShutdownHook(false)
.parent(parentContext);
}
static void prepareSharedChannelRegistry(SharedChannelRegistry sharedChannelRegistry,
LinkedHashMap<Class<?>, String> appsWithNamespace) {
int i = 0;
SubscribableChannel sharedChannel = null;
for (Entry<Class<?>, String> appEntry : appsWithNamespace.entrySet()) {
String namespace = appEntry.getValue();
if (i > 0) {
sharedChannelRegistry.register(namespace + "." + INPUT_CHANNEL_NAME, sharedChannel);
}
sharedChannel = new DirectChannel();
if (i < appsWithNamespace.size() - 1) {
sharedChannelRegistry.register(namespace + "." + OUTPUT_CHANNEL_NAME, sharedChannel);
}
i++;
}
}
public interface AggregateApplication {
/**
* Retrieves the bindable proxy instance (e.g. {@link org.springframework.cloud.stream.messaging.Processor},
* {@link org.springframework.cloud.stream.messaging.Source},
* {@link org.springframework.cloud.stream.messaging.Sink} or custom interface) from
* the given namespace.
*
* @param bindableType the bindable type
* @param namespace the namespace
* @param <T>
* @return
*/
<T> T getBinding(Class<T> bindableType, String namespace);
}

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

@@ -26,6 +26,9 @@ import java.util.List;
import java.util.Map;
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.endpoint.MetricReaderPublicMetrics;
import org.springframework.boot.actuate.endpoint.MetricsEndpoint;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
@@ -35,6 +38,9 @@ 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.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.core.env.PropertySources;
@@ -49,7 +55,8 @@ import org.springframework.util.StringUtils;
* @author Marius Bogoevici
* @author Venil Noronha
*/
public class AggregateApplicationBuilder {
@EnableBinding
public class AggregateApplicationBuilder implements AggregateApplication, ApplicationContextAware, SmartInitializingSingleton {
private SourceConfigurer sourceConfigurer;
@@ -127,15 +134,38 @@ public class AggregateApplicationBuilder {
return this;
}
@Override
public void afterSingletonsInstantiated() {
this.run();
}
@Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
this.parentContext = (ConfigurableApplicationContext) applicationContext;
}
@Override
public <T> T getBinding(Class<T> bindableType, String namespace) {
if (parentContext == null) {
throw new IllegalStateException("The aggregate application has not been started yet");
}
try {
return bindableType.cast(parentContext.getBean(namespace + "." + bindableType.getName()));
} catch (BeansException e) {
throw new IllegalStateException("Binding not found for '" + bindableType.getName() + "' into namespace " +
namespace);
}
}
public SourceConfigurer from(Class<?> app) {
SourceConfigurer sourceConfigurer = new SourceConfigurer(app);
this.sourceConfigurer = sourceConfigurer;
return sourceConfigurer;
}
public ConfigurableApplicationContext run(String[] parentArgsFromRun) {
this.parentArgs.addAll(Arrays.asList(parentArgsFromRun));
List<AppConfigurer<?>> apps = new ArrayList<AppConfigurer<?>>();
public ConfigurableApplicationContext run(String... parentArgs) {
this.parentArgs.addAll(Arrays.asList(parentArgs));
List<AppConfigurer<?>> apps = new ArrayList<>();
if (this.sourceConfigurer != null) {
apps.add(sourceConfigurer);
}
@@ -154,16 +184,25 @@ public class AggregateApplicationBuilder {
Class<?> appToEmbed = appConfigurer.getApp();
// Always update namespace before preparing SharedChannelRegistry
if (appConfigurer.namespace == null) {
appConfigurer.namespace = AggregateApplication.getDefaultNamespace(appConfigurer.getApp().getName(),
appConfigurer.namespace = AggregateApplicationUtils.getDefaultNamespace(appConfigurer.getApp().getName(),
i);
}
appsToEmbed.put(appToEmbed, appConfigurer.namespace);
appConfigurers.put(appConfigurer, appConfigurer.namespace);
}
this.parentContext = AggregateApplication.createParentContext(this.parentSources.toArray(new Object[0]),
if (this.parentContext == null) {
this.parentContext = AggregateApplicationUtils.createParentContext(this.parentSources.toArray(new Object[0]),
this.parentArgs.toArray(new String[0]), selfContained(), this.webEnvironment, this.headless);
}
else {
if (BeanFactoryUtils.beansOfTypeIncludingAncestors(this.parentContext, SharedChannelRegistry.class)
.size() == 0) {
this.parentContext.getBeanFactory().registerSingleton("sharedChannelRegistry",
new SharedChannelRegistry());
}
}
SharedChannelRegistry sharedChannelRegistry = this.parentContext.getBean(SharedChannelRegistry.class);
AggregateApplication.prepareSharedChannelRegistry(sharedChannelRegistry, appsToEmbed);
AggregateApplicationUtils.prepareSharedChannelRegistry(sharedChannelRegistry, appsToEmbed);
PropertySources propertySources = this.parentContext.getEnvironment()
.getPropertySources();
for (Map.Entry<AppConfigurer, String> appConfigurerEntry : appConfigurers
@@ -209,6 +248,10 @@ public class AggregateApplicationBuilder {
AppConfigurer<?> appConfigurer = apps.get(i);
appConfigurer.embed();
}
if (BeanFactoryUtils.beansOfTypeIncludingAncestors(this.parentContext, AggregateApplication.class)
.size() == 0) {
this.parentContext.getBeanFactory().registerSingleton("aggregateApplicationAccessor", this);
}
return this.parentContext;
}
@@ -228,7 +271,7 @@ public class AggregateApplicationBuilder {
private ChildContextBuilder childContext(Class<?> app,
ConfigurableApplicationContext parentContext, String namespace) {
return new ChildContextBuilder(AggregateApplication.embedApp(parentContext,
return new ChildContextBuilder(AggregateApplicationUtils.embedApp(parentContext,
namespace, app));
}
@@ -325,13 +368,38 @@ public class AggregateApplicationBuilder {
childContext(this.app, AggregateApplicationBuilder.this.parentContext,
this.namespace).args(this.args).config(this.names)
.profiles(this.profiles).run();
AggregateApplicationBuilder.this.parentContext.getBeanFactory().getBean(
MetricsEndpoint.class).registerPublicMetrics(
new MetricReaderPublicMetrics(
new NamespaceAwareSpringIntegrationMetricReader(this.namespace, childContext.getBean(
IntegrationMBeanExporter.class))));
// Register bindable proxies as beans so they can be queried for later
Map<String, BindableProxyFactory> bindableProxies = BeanFactoryUtils
.beansOfTypeIncludingAncestors(childContext.getBeanFactory(), BindableProxyFactory.class);
for (String bindableProxyName : bindableProxies.keySet()) {
try {
AggregateApplicationBuilder.this.parentContext.getBeanFactory().registerSingleton(
this.getNamespace() + "." + bindableProxyName.substring(1, bindableProxyName.length()),
bindableProxies.get(bindableProxyName).getObject());
}
catch (Exception e) {
throw new IllegalStateException(
"Error while trying to register the aggregate bound interface '"
+ bindableProxyName + "' into namespace '" + this.getNamespace() + "'",
e);
}
}
// Register metrics if JMX enabled and exporter avalable
if (BeanFactoryUtils.beansOfTypeIncludingAncestors(AggregateApplicationBuilder.this.parentContext,
IntegrationMBeanExporter.class).size() > 0) {
BeanFactoryUtils
.beanOfTypeIncludingAncestors(AggregateApplicationBuilder.this.parentContext, MetricsEndpoint.class)
.registerPublicMetrics(
new MetricReaderPublicMetrics(new NamespaceAwareSpringIntegrationMetricReader(
this.namespace, childContext.getBean(IntegrationMBeanExporter.class))));
}
}
public AggregateApplication build() {
return applicationBuilder;
}
public String[] getArgs() {
return this.args;
}
@@ -394,4 +462,5 @@ public class AggregateApplicationBuilder {
return new SharedChannelRegistry();
}
}
}

98
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/AggregateApplicationUtils.java Normal file
View File

@@ -0,0 +1,98 @@
/*
* Copyright 2015-2016 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.aggregate;
import java.util.LinkedHashMap;
import java.util.Map.Entry;
import org.springframework.boot.Banner.Mode;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.messaging.SubscribableChannel;
/**
* Utilities for embedding applications in aggregates.
*
* @author Marius Bogoevici
* @author Ilayaperumal Gopinathan
* @author Venil Noronha
*/
abstract class AggregateApplicationUtils {
private static final String SPRING_CLOUD_STREAM_INTERNAL_PREFIX = "spring.cloud.stream.internal";
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";
static ConfigurableApplicationContext createParentContext(Object[] sources, String[] args, final
boolean selfContained, boolean webEnvironment,
boolean headless) {
SpringApplicationBuilder aggregatorParentConfiguration = new SpringApplicationBuilder();
aggregatorParentConfiguration
.sources(sources)
.web(webEnvironment)
.headless(headless)
.properties("spring.jmx.default-domain="
+ AggregateApplicationBuilder.ParentConfiguration.class.getName(),
SELF_CONTAINED_APP_PROPERTY_NAME + "=" + selfContained);
return aggregatorParentConfiguration.run(args);
}
static String getDefaultNamespace(String appClassName, int index) {
return appClassName + "_" + index;
}
protected static SpringApplicationBuilder embedApp(
ConfigurableApplicationContext parentContext, String namespace,
Class<?> app) {
return new SpringApplicationBuilder(app)
.web(false)
.main(app)
.bannerMode(Mode.OFF)
.properties("spring.jmx.default-domain=" + namespace)
.properties(CHANNEL_NAMESPACE_PROPERTY_NAME + "=" + namespace)
.registerShutdownHook(false)
.parent(parentContext);
}
static void prepareSharedChannelRegistry(SharedChannelRegistry sharedChannelRegistry,
LinkedHashMap<Class<?>, String> appsWithNamespace) {
int i = 0;
SubscribableChannel sharedChannel = null;
for (Entry<Class<?>, String> appEntry : appsWithNamespace.entrySet()) {
String namespace = appEntry.getValue();
if (i > 0) {
sharedChannelRegistry.register(namespace + "." + INPUT_CHANNEL_NAME, sharedChannel);
}
sharedChannel = new DirectChannel();
if (i < appsWithNamespace.size() - 1) {
sharedChannelRegistry.register(namespace + "." + OUTPUT_CHANNEL_NAME, sharedChannel);
}
i++;
}
}
}

8
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/aggregation/ModuleAggregationTest.java → spring-cloud-stream/src/test/java/org/springframework/cloud/stream/aggregation/AggregationTest.java
View File

@@ -44,7 +44,7 @@ import static org.junit.Assert.assertTrue;
* @author Marius Bogoevici
* @author Ilayaperumal Gopinathan
*/
public class ModuleAggregationTest {
public class AggregationTest {
private ConfigurableApplicationContext aggregatedApplicationContext;
@@ -60,10 +60,12 @@ public class ModuleAggregationTest {
}
@Test
public void testModuleAggregation() {
public void aggregation() {
aggregatedApplicationContext = new AggregateApplicationBuilder(
MockBinderRegistryConfiguration.class, "--server.port=0")
.from(TestSource.class).to(TestProcessor.class).run();
.from(TestSource.class)
.to(TestProcessor.class)
.run();
SharedChannelRegistry sharedChannelRegistry = aggregatedApplicationContext
.getBean(SharedChannelRegistry.class);
BindableChannelFactory channelFactory = aggregatedApplicationContext

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

@@ -25,7 +25,8 @@ import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* A simple configuration that creates mock {@link org.springframework.cloud.stream.binder.Binder}s.
* A simple configuration that creates mock
* {@link org.springframework.cloud.stream.binder.Binder}s.
* @author Marius Bogoevici
*/
@Configuration