diff --git a/spring-cloud-stream-test-support/src/test/java/org/springframework/cloud/stream/test/aggregate/AggregateTestWithBean.java b/spring-cloud-stream-test-support/src/test/java/org/springframework/cloud/stream/test/aggregate/AggregateTestWithBean.java new file mode 100644 index 000000000..aa7aee1b8 --- /dev/null +++ b/spring-cloud-stream-test-support/src/test/java/org/springframework/cloud/stream/test/aggregate/AggregateTestWithBean.java @@ -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!"; + } + } +} 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 new file mode 100644 index 000000000..83f245605 --- /dev/null +++ b/spring-cloud-stream-test-support/src/test/java/org/springframework/cloud/stream/test/aggregate/AggregateTestWithMain.java @@ -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!"; + } + } +} diff --git a/spring-cloud-stream-test-support/src/test/java/org/springframework/cloud/stream/test/ExampleTest.java b/spring-cloud-stream-test-support/src/test/java/org/springframework/cloud/stream/test/example/ExampleTest.java similarity index 98% rename from spring-cloud-stream-test-support/src/test/java/org/springframework/cloud/stream/test/ExampleTest.java rename to spring-cloud-stream-test-support/src/test/java/org/springframework/cloud/stream/test/example/ExampleTest.java index 61ede5f37..7de0b9462 100644 --- a/spring-cloud-stream-test-support/src/test/java/org/springframework/cloud/stream/test/ExampleTest.java +++ b/spring-cloud-stream-test-support/src/test/java/org/springframework/cloud/stream/test/example/ExampleTest.java @@ -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; 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 e56e42e39..7945d4de8 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 @@ -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, String> appsWithNamespace) { - int i = 0; - SubscribableChannel sharedChannel = null; - for (Entry, 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 + * @return + */ + T getBinding(Class bindableType, String namespace); } 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 99102f15d..600f02742 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 @@ -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 getBinding(Class 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> apps = new ArrayList>(); + public ConfigurableApplicationContext run(String... parentArgs) { + this.parentArgs.addAll(Arrays.asList(parentArgs)); + List> 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 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 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(); } } + } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/AggregateApplicationUtils.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/AggregateApplicationUtils.java new file mode 100644 index 000000000..58f1b0620 --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/AggregateApplicationUtils.java @@ -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, String> appsWithNamespace) { + int i = 0; + SubscribableChannel sharedChannel = null; + for (Entry, 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++; + } + } + +} diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/aggregation/ModuleAggregationTest.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/aggregation/AggregationTest.java similarity index 99% rename from spring-cloud-stream/src/test/java/org/springframework/cloud/stream/aggregation/ModuleAggregationTest.java rename to spring-cloud-stream/src/test/java/org/springframework/cloud/stream/aggregation/AggregationTest.java index 73086bd31..a47595035 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/aggregation/ModuleAggregationTest.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/aggregation/AggregationTest.java @@ -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 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 3f9ba5850..8fc7da962 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 @@ -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