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

Add support for direct binding

Browse Source 
and refactor channel proxy behaviour

- Remove DirectChannelProxyFactory;
- Proxies create channels as necessary, including support for PollableChannels;
- Channel beans are using the bean factory methods for the proxies;
- Coordinate bind/unbind across a context through the internal Bindable interface;
- Add support for embedding modules as independent child contexts of the main module, wrapped in a Bindable interface;
- Add support for channel name namespacing;
- Use a shared channel registry (consulted by the proxies) to create direct bindings;

Restrict the use of SharedChannelRegistry to aggregation

Fixed copyrights

Polishing based on comments from the previous branch

Tweaks

- namespacing takes into account module index, so multiple modules with the same class can partiticipate
- fix how properties are passes to parent/child
- Split ChannelBindingLifecycle from ChannelBindingAdapter and register it with each child;

Javadoc updates

Pass only  prefixed properties

Actually use the filtering method

Lifecycle simplifications

- Removing BindableChannelWrapper and UnbindOnCloseApplicationListener and relying on ChannelBindingListener solely;
- Some code cleanup

Further simplifications

More cleanup

Remove BindableUtils, rename ChannelBindingAdapter

Minor improvements

Addressing some comments, renaming, etc

More polishing
This commit is contained in:
Marius Bogoevici
2015-08-06 09:04:57 -04:00
committed by Mark Fisher
parent 80391fa876
commit bd47560bd5
55 changed files with 1368 additions and 1451 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.gitignore vendored
View File

@@ -21,3 +21,5 @@ _site/
.factorypath
spring-xd-samples/*/xd
dump.rdb
.apt_generated
artifacts

40
spring-cloud-stream-binders/spring-cloud-stream-binder-local/src/main/java/org/springframework/cloud/stream/binder/local/config/LocalExecutorConfigurationProperties.java
View File

@@ -24,43 +24,43 @@ import org.springframework.boot.context.properties.ConfigurationProperties;
@ConfigurationProperties(prefix = "spring.cloud.stream.binder.local.executor")
class LocalExecutorConfigurationProperties {
private int executorCorePoolSize;
private int corePoolSize;
private int executorMaxPoolSize;
private int maxPoolSize;
private int executorQueueSize = Integer.MAX_VALUE;
private int queueSize = Integer.MAX_VALUE;
private int executorKeepAliveSeconds;
private int keepAliveSeconds;
public int getExecutorCorePoolSize() {
return executorCorePoolSize;
public int getCorePoolSize() {
return corePoolSize;
}
public void setExecutorCorePoolSize(int executorCorePoolSize) {
this.executorCorePoolSize = executorCorePoolSize;
public void setCorePoolSize(int corePoolSize) {
this.corePoolSize = corePoolSize;
}
public int getExecutorMaxPoolSize() {
return executorMaxPoolSize;
public int getMaxPoolSize() {
return maxPoolSize;
}
public void setExecutorMaxPoolSize(int executorMaxPoolSize) {
this.executorMaxPoolSize = executorMaxPoolSize;
public void setMaxPoolSize(int maxPoolSize) {
this.maxPoolSize = maxPoolSize;
}
public int getExecutorQueueSize() {
return executorQueueSize;
public int getQueueSize() {
return queueSize;
}
public void setExecutorQueueSize(int executorQueueSize) {
this.executorQueueSize = executorQueueSize;
public void setQueueSize(int queueSize) {
this.queueSize = queueSize;
}
public int getExecutorKeepAliveSeconds() {
return executorKeepAliveSeconds;
public int getKeepAliveSeconds() {
return keepAliveSeconds;
}
public void setExecutorKeepAliveSeconds(int executorKeepAliveSeconds) {
this.executorKeepAliveSeconds = executorKeepAliveSeconds;
public void setKeepAliveSeconds(int keepAliveSeconds) {
this.keepAliveSeconds = keepAliveSeconds;
}
}

8
spring-cloud-stream-binders/spring-cloud-stream-binder-local/src/main/java/org/springframework/cloud/stream/binder/local/config/LocalMessageChannelBinderConfiguration.java
View File

@@ -44,10 +44,10 @@ public class LocalMessageChannelBinderConfiguration {
public LocalMessageChannelBinder localMessageChannelBinder() {
LocalMessageChannelBinder localMessageChannelBinder = new LocalMessageChannelBinder();
localMessageChannelBinder.setExecutorCorePoolSize(localExecutorConfigurationProperties.getExecutorCorePoolSize());
localMessageChannelBinder.setExecutorKeepAliveSeconds(localExecutorConfigurationProperties.getExecutorKeepAliveSeconds());
localMessageChannelBinder.setExecutorMaxPoolSize(localExecutorConfigurationProperties.getExecutorMaxPoolSize());
localMessageChannelBinder.setExecutorQueueSize(localExecutorConfigurationProperties.getExecutorQueueSize());
localMessageChannelBinder.setExecutorCorePoolSize(localExecutorConfigurationProperties.getCorePoolSize());
localMessageChannelBinder.setExecutorKeepAliveSeconds(localExecutorConfigurationProperties.getKeepAliveSeconds());
localMessageChannelBinder.setExecutorMaxPoolSize(localExecutorConfigurationProperties.getMaxPoolSize());
localMessageChannelBinder.setExecutorQueueSize(localExecutorConfigurationProperties.getQueueSize());
if (polling > 0) {
PollerMetadata pollerMetadata = new PollerMetadata();

1
spring-cloud-stream-module-launcher/.gitignore vendored
View File

@@ -1 +0,0 @@
artifacts

4
spring-cloud-stream-module-launcher/src/main/java/org/springframework/boot/loader/ModuleJarLauncher.java
View File

@@ -84,6 +84,7 @@ public class ModuleJarLauncher extends ExecutableArchiveLauncher {
if (systemClassLoader instanceof URLClassLoader) {
// add the URLs of the application classloader to the created classloader
// to compensate for LaunchedURLClassLoader not delegating to parent to retrieve resources
@SuppressWarnings("resource")
URLClassLoader systemUrlClassLoader = (URLClassLoader) systemClassLoader;
URL[] mergedUrls = new URL[urls.length + systemUrlClassLoader.getURLs().length];
System.arraycopy(urls, 0, mergedUrls, 0, urls.length);
@@ -91,9 +92,8 @@ public class ModuleJarLauncher extends ExecutableArchiveLauncher {
systemUrlClassLoader.getURLs().length);
// add the extension classloader as parent to the created context, if accessible
return new LaunchedURLClassLoader(mergedUrls, systemUrlClassLoader.getParent());
} else {
return new LaunchedURLClassLoader(urls, systemClassLoader);
}
return new LaunchedURLClassLoader(urls, systemClassLoader);
}
}

39
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/adapter/ChannelBinding.java → spring-cloud-stream-module-launcher/src/main/java/org/springframework/cloud/stream/module/launcher/ArchiveMatchingEntryFilter.java
View File

@@ -14,37 +14,26 @@
* limitations under the License.
*/
package org.springframework.cloud.stream.adapter;
package org.springframework.cloud.stream.module.launcher;
import org.springframework.boot.loader.archive.Archive;
import org.springframework.boot.loader.util.AsciiBytes;
/**
* Represents a binding between a local and remote message channel.
*
* @author Dave Syer
* @author Mark Fisher
* @author Marius Bogoevici
*/
public abstract class ChannelBinding {
class ArchiveMatchingEntryFilter implements Archive.EntryFilter {
private String localName;
private String remoteName;
public static final ArchiveMatchingEntryFilter FILTER = new ArchiveMatchingEntryFilter();
protected ChannelBinding() {
this(null);
private static final AsciiBytes LIB = new AsciiBytes("lib/");
@Override
public boolean matches(Archive.Entry entry) {
return isNestedArchive(entry);
}
protected ChannelBinding(String localName) {
this.localName = localName;
protected boolean isNestedArchive(Archive.Entry entry) {
return !entry.isDirectory() && entry.getName().startsWith(LIB);
}
public String getLocalName() {
return this.localName;
}
public String getRemoteName() {
return this.remoteName;
}
public void setRemoteName(String name) {
this.remoteName = name;
}
}

122
spring-cloud-stream-module-launcher/src/main/java/org/springframework/cloud/stream/module/launcher/ModuleLauncher.java
View File

@@ -17,6 +17,8 @@
package org.springframework.cloud.stream.module.launcher;
import java.io.IOException;
import java.lang.reflect.Method;
import java.net.URL;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
@@ -24,11 +26,19 @@ import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.boot.loader.LaunchedURLClassLoader;
import org.springframework.boot.loader.ModuleJarLauncher;
import org.springframework.boot.loader.archive.Archive;
import org.springframework.boot.loader.archive.JarFileArchive;
import org.springframework.cloud.stream.module.resolver.ModuleResolver;
import org.springframework.cloud.stream.module.utils.ClassloaderUtils;
import org.springframework.core.io.Resource;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import org.springframework.util.ReflectionUtils;
import org.springframework.util.StringUtils;
/**
@@ -42,6 +52,14 @@ import org.springframework.util.StringUtils;
*/
public class ModuleLauncher {
public static final String AGGREGATE_APPLICATION_CLASS = "org.springframework.cloud.stream.aggregate.AggregateApplication";
public static final String AGGREGATE_APPLICATION_RUN_METHOD = "run";
public static final String MODULE_AGGREGATOR_RUNNER_THREAD_NAME = "module-aggregator-runner";
private Log log = LogFactory.getLog(ModuleLauncher.class);
private static final String DEFAULT_EXTENSION = "jar";
private static final String DEFAULT_CLASSIFIER = "exec";
@@ -68,15 +86,26 @@ public class ModuleLauncher {
* <code>&lt;groupId&gt;:&lt;artifactId&gt;[:&lt;extension&gt;[:&lt;classifier&gt;]]:&lt;version&gt;</code>
*
* @param moduleLaunchRequests a list of modules with their arguments
* @param aggregate whether the modules should be aggregated at launch
* @param parentArgs a list of arguments for the whole aggregate
*/
public void launch(List<ModuleLaunchRequest> moduleLaunchRequests) {
public void launch(List<ModuleLaunchRequest> moduleLaunchRequests, boolean aggregate, String parentArgs[]) {
List<ModuleLaunchRequest> reversed = new ArrayList<>(moduleLaunchRequests);
Collections.reverse(reversed);
for (ModuleLaunchRequest moduleLaunchRequest : reversed) {
String module = moduleLaunchRequest.getModule();
moduleLaunchRequest.addArgument("spring.jmx.default-domain", module.replace("/", ".").replace(":", "."));
launchModule(module, toArgArray(moduleLaunchRequest.getArguments()));
if (moduleLaunchRequests.size() == 1 || !aggregate) {
launchIndividualModules(moduleLaunchRequests);
}
else {
launchAggregatedModules(moduleLaunchRequests, parentArgs);
}
}
public void launch(List<ModuleLaunchRequest> moduleLaunchRequests, boolean aggregate) {
this.launch(moduleLaunchRequests, aggregate, new String[0]);
}
public void launch(List<ModuleLaunchRequest> moduleLaunchRequests) {
this.launch(moduleLaunchRequests, false);
}
/**
@@ -92,6 +121,56 @@ public class ModuleLauncher {
return result;
}
public void launchAggregatedModules(List<ModuleLaunchRequest> moduleLaunchRequests, final String[] parentArgs) {
try {
List<String> mainClassNames = new ArrayList<>();
List<URL> jarURLs = new ArrayList<>();
List<String> seenArchives = new ArrayList<>();
final List<String[]> arguments = new ArrayList<>();
// aggregate jars from all modules and extract their main Classes
for (ModuleLaunchRequest moduleLaunchRequest : moduleLaunchRequests) {
Resource resource = resolveModule(moduleLaunchRequest.getModule());
JarFileArchive jarFileArchive = new JarFileArchive(resource.getFile());
jarURLs.add(jarFileArchive.getUrl());
for (Archive archive : jarFileArchive.getNestedArchives(ArchiveMatchingEntryFilter.FILTER)) {
// avoid duplication based on unique JAR names
// TODO - read the metadata from the JARs, do proper version resolution on merge
String urlAsString = archive.getUrl().toString();
String jarNameWithExtension = urlAsString.substring(0, urlAsString.lastIndexOf("!/"));
String jarNameWithoutExtension = jarNameWithExtension.substring(jarNameWithExtension.lastIndexOf("/") + 1);
if (!seenArchives.contains(jarNameWithoutExtension)) {
seenArchives.add(jarNameWithoutExtension);
jarURLs.add(archive.getUrl());
}
}
mainClassNames.add(jarFileArchive.getMainClass());
arguments.add(toArgArray(moduleLaunchRequest.getArguments()));
}
final ClassLoader classLoader = new LaunchedURLClassLoader(jarURLs.toArray(new URL[jarURLs.size()]),
ClassloaderUtils.getExtensionClassloader());
final List<Class<?>> mainClasses = new ArrayList<>();
for (String mainClass : mainClassNames) {
mainClasses.add(ClassUtils.forName(mainClass, classLoader));
}
Runnable moduleAggregatorRunner = new ModuleAggregatorRunner(classLoader, mainClasses, parentArgs, arguments);
Thread moduleAggregatorRunnerThread = new Thread(moduleAggregatorRunner);
moduleAggregatorRunnerThread.setContextClassLoader(classLoader);
moduleAggregatorRunnerThread.setName(MODULE_AGGREGATOR_RUNNER_THREAD_NAME);
moduleAggregatorRunnerThread.start();
} catch (Exception e) {
throw new RuntimeException("failed to start aggregated modules: " + StringUtils.collectionToCommaDelimitedString(moduleLaunchRequests), e);
}
}
public void launchIndividualModules(List<ModuleLaunchRequest> reversed) {
for (ModuleLaunchRequest moduleLaunchRequest : reversed) {
String module = moduleLaunchRequest.getModule();
moduleLaunchRequest.addArgument("spring.jmx.default-domain", module.replace("/", ".").replace(":", "."));
launchModule(module, toArgArray(moduleLaunchRequest.getArguments()));
}
}
private void launchModule(String module, String[] args) {
try {
Resource resource = resolveModule(module);
@@ -116,4 +195,37 @@ public class ModuleLauncher {
return this.moduleResolver.resolve(groupId, artifactId, extension, classifier, version);
}
private class ModuleAggregatorRunner implements Runnable {
private final ClassLoader classLoader;
private final String[] parentArgs;
private final List<Class<?>> mainClasses;
private final List<String[]> arguments;
public ModuleAggregatorRunner(ClassLoader classLoader, List<Class<?>> mainClasses, String[] parentArgs, List<String[]> moduleArguments) {
this.classLoader = classLoader;
this.parentArgs = parentArgs;
this.mainClasses = mainClasses;
this.arguments = moduleArguments;
}
@Override
public void run() {
try {
// we expect the class and method to be found on the module classpath
Class<?> moduleAggregatorClass = ClassUtils.forName(AGGREGATE_APPLICATION_CLASS, classLoader);
Method aggregateMethod = ReflectionUtils.findMethod(moduleAggregatorClass,
AGGREGATE_APPLICATION_RUN_METHOD, Class[].class, String[].class, String[][].class);
aggregateMethod.invoke(null,
mainClasses.toArray(new Class<?>[mainClasses.size()]),
parentArgs, arguments.toArray(new String[][] {}));
} catch (Exception e) {
log.error("failed to launch aggregated modules :" + StringUtils.collectionToCommaDelimitedString(mainClasses), e);
throw new RuntimeException(e);
}
}
}
}

17
spring-cloud-stream-module-launcher/src/main/java/org/springframework/cloud/stream/module/launcher/ModuleLauncherProperties.java
View File

@@ -49,12 +49,17 @@ import org.springframework.core.env.Environment;
public class ModuleLauncherProperties {
/**
* Array of coordinates for modules that need to be launched.
* True if aggregating multiple modules when launched together
*/
private boolean aggregate;
/**
* File path to a locally available maven repository, where modules will be downloaded.
*/
private String[] modules;
/**
* Map of arguments, keyed by the 0-based index in the {@kink #modules array}.
* Map of arguments, keyed by the 0-based index in the {@link #modules array}.
*/
private Map<Integer, Map<String, String>> args = new HashMap<>();
@@ -62,6 +67,14 @@ public class ModuleLauncherProperties {
this.modules = modules;
}
public boolean isAggregate() {
return aggregate;
}
public void setAggregate(boolean aggregate) {
this.aggregate = aggregate;
}
@NotEmpty(message = "A list of modules must be specified.")
public String[] getModules() {
return modules;

15
spring-cloud-stream-module-launcher/src/main/java/org/springframework/cloud/stream/module/launcher/ModuleLauncherRunner.java
View File

@@ -58,7 +58,9 @@ public class ModuleLauncherRunner implements CommandLineRunner {
}
log.info(sb.toString());
}
this.moduleLauncher.launch(launchRequests);
this.moduleLauncher.launch(launchRequests,
moduleLauncherProperties.isAggregate(),
moduleLauncherProperties.isAggregate() ? extractAggregateProperties(args) : new String[0]);
}
private List<ModuleLaunchRequest> generateModuleLaunchRequests() {
@@ -71,4 +73,15 @@ public class ModuleLauncherRunner implements CommandLineRunner {
}
return requests;
}
private String[] extractAggregateProperties(String[] args) {
List<String> filteredProperties = new ArrayList<>();
for (String arg : args) {
if (arg.startsWith("--spring.")) {
filteredProperties.add(arg);
}
}
return filteredProperties.toArray(new String[filteredProperties.size()]);
}
}

2
spring-cloud-stream-module-launcher/src/main/java/org/springframework/cloud/stream/module/resolver/ModuleResolver.java
View File

@@ -36,6 +36,6 @@ public interface ModuleResolver {
* @param version the version
* @return the resource
*/
public Resource resolve(String groupId, String artifactId, String extension, String classifier, String version);
Resource resolve(String groupId, String artifactId, String extension, String classifier, String version);
}

2
spring-cloud-stream-module-launcher/src/main/java/org/springframework/cloud/stream/module/resolver/ModuleResolverProperties.java
View File

@@ -5,7 +5,7 @@
* 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
* 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,

39
spring-cloud-stream-module-launcher/src/main/java/org/springframework/cloud/stream/module/utils/ClassloaderUtils.java Normal file
View File

@@ -0,0 +1,39 @@
/*
* Copyright 2015 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.module.utils;
/**
* @author Marius Bogoevici
*/
public class ClassloaderUtils {
/**
* Retrieves the extension classloader of the current JVM, if accessible. In general the extension classloader is
* found in a hierarchy as the parent of the application classloader. If such a hierarchy does not exist, it will
* return the application classloader itself.
*
* @return the classloader
*/
public static ClassLoader getExtensionClassloader() {
ClassLoader systemClassLoader = ClassLoader.getSystemClassLoader();
// try to retrieve the extension classloader
ClassLoader extensionClassLoader = systemClassLoader != null ? systemClassLoader.getParent() : null;
// set the classloader for the module as the extension classloader if available
// fall back to the system classloader (which can also be null) if not available
return extensionClassLoader != null ? extensionClassLoader : systemClassLoader;
}
}

15
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/adapter/ChannelLocator.java → spring-cloud-stream-samples/double/src/main/java/config/sink/SinkApplication.java
View File

@@ -14,16 +14,13 @@
* limitations under the License.
*/
package org.springframework.cloud.stream.adapter;
package config.sink;
import org.springframework.boot.autoconfigure.SpringBootApplication;
/**
* @author Dave Syer
*
* @author Marius Bogoevici
*/
public interface ChannelLocator {
String locate(String name);
String tap(String name);
@SpringBootApplication
public class SinkApplication {
}

2
spring-cloud-stream-samples/double/src/main/java/config/SinkModuleDefinition.java → spring-cloud-stream-samples/double/src/main/java/config/sink/SinkModuleDefinition.java
View File

@@ -14,7 +14,7 @@
* limitations under the License.
*/
package config;
package config.sink;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

18
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/adapter/InputChannelBinding.java → spring-cloud-stream-samples/double/src/main/java/config/source/SourceApplication.java
View File

@@ -14,19 +14,13 @@
* limitations under the License.
*/
package org.springframework.cloud.stream.adapter;
package config.source;
import org.springframework.boot.autoconfigure.SpringBootApplication;
/**
* @author Dave Syer
* @author Marius Bogoevici
*/
public class InputChannelBinding extends ChannelBinding {
protected InputChannelBinding() {
super(null);
}
public InputChannelBinding(String localName) {
super(localName);
}
@SpringBootApplication
public class SourceApplication {
}

7
spring-cloud-stream-samples/double/src/main/java/config/SourceModuleDefinition.java → spring-cloud-stream-samples/double/src/main/java/config/source/SourceModuleDefinition.java
View File

@@ -14,7 +14,7 @@
* limitations under the License.
*/
package config;
package config.source;
import java.text.SimpleDateFormat;
import java.util.Date;
@@ -35,11 +35,10 @@ import org.springframework.messaging.support.GenericMessage;
@EnableModule(Source.class)
public class SourceModuleDefinition {
@Value("${format:YYYY/MM/dd hh:mm:ss}")
private String format;
private String format = "yyyy-MM-dd HH:mm:ss";
@Bean
@InboundChannelAdapter(value = Source.OUTPUT, autoStartup = "false", poller = @Poller(fixedDelay = "${fixedDelay}", maxMessagesPerPoll = "1"))
@InboundChannelAdapter(value = Source.OUTPUT, poller = @Poller(fixedDelay = "${fixedDelay}", maxMessagesPerPoll = "1"))
public MessageSource<String> timerMessageSource() {
return () -> new GenericMessage<>(new SimpleDateFormat(this.format).format(new Date()));
}

18
spring-cloud-stream-samples/double/src/main/java/demo/DoubleApplication.java
View File

@@ -16,25 +16,17 @@
package demo;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.stream.aggregate.AggregateBuilder;
import org.springframework.cloud.stream.aggregate.AggregateConfigurer;
import org.springframework.cloud.stream.aggregate.AggregateApplication;
import config.SinkModuleDefinition;
import config.SourceModuleDefinition;
import config.sink.SinkApplication;
import config.source.SourceApplication;
@SpringBootApplication
public class DoubleApplication implements AggregateConfigurer {
@Override
public void configure(AggregateBuilder builder) {
builder.from(SourceModuleDefinition.class).as("source")
.to(SinkModuleDefinition.class).as("sink");
}
public class DoubleApplication {
public static void main(String[] args) {
SpringApplication.run(DoubleApplication.class, args);
AggregateApplication.run(SourceApplication.class, SinkApplication.class);
}
}

2
spring-cloud-stream-samples/double/src/main/resources/application.yml
View File

@@ -1 +1 @@
fixedDelay: 1000

425
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/adapter/ChannelBindingAdapter.java
View File

@@ -1,425 +0,0 @@
/*
* Copyright 2015 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.adapter;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.atomic.AtomicBoolean;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.cloud.stream.binder.Binder;
import org.springframework.cloud.stream.binder.BinderHeaders;
import org.springframework.cloud.stream.config.ChannelBindingProperties;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.SmartLifecycle;
import org.springframework.integration.channel.ChannelInterceptorAware;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.interceptor.WireTap;
import org.springframework.integration.support.DefaultMessageBuilderFactory;
import org.springframework.integration.support.MessageBuilderFactory;
import org.springframework.jmx.export.annotation.ManagedAttribute;
import org.springframework.jmx.export.annotation.ManagedOperation;
import org.springframework.jmx.export.annotation.ManagedResource;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.core.DestinationResolver;
import org.springframework.messaging.support.ChannelInterceptorAdapter;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
/**
* Binds input/output channels.
*
* @author Mark Fisher
* @author Dave Syer
* @author Marius Bogoevici
* @author Ilayaperumal Gopinathan
*/
@ManagedResource
public class ChannelBindingAdapter implements SmartLifecycle, ApplicationContextAware {
private static Logger logger = LoggerFactory.getLogger(ChannelBindingAdapter.class);
private Binder<MessageChannel> binder;
private MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory();
private Collection<OutputChannelBinding> outputChannels = Collections.emptySet();
private Collection<InputChannelBinding> inputChannels = Collections.emptySet();
private boolean running = false;
private final AtomicBoolean active = new AtomicBoolean(false);
private boolean trackHistory = false;
private ChannelBindingProperties module;
private ConfigurableApplicationContext applicationContext;
private ChannelLocator channelLocator;
private DestinationResolver<MessageChannel> channelResolver;
private Map<String, String> bindings = new HashMap<String, String>();
public ChannelBindingAdapter(ChannelBindingProperties module, Binder<MessageChannel> binder) {
this.module = module;
this.binder = binder;
}
public void setChannelLocator(ChannelLocator channelLocator) {
this.channelLocator = channelLocator;
}
@Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
this.applicationContext = (ConfigurableApplicationContext) applicationContext;
}
public void setChannelResolver(DestinationResolver<MessageChannel> channelResolver) {
this.channelResolver = channelResolver;
}
public void setMessageBuilderFactory(MessageBuilderFactory messageBuilderFactory) {
this.messageBuilderFactory = messageBuilderFactory;
}
public void setTrackHistory(boolean trackHistory) {
this.trackHistory = trackHistory;
}
public void setOutputChannels(Collection<OutputChannelBinding> outputChannels) {
this.outputChannels = new LinkedHashSet<OutputChannelBinding>(outputChannels);
}
public void setInputChannels(Collection<InputChannelBinding> inputChannels) {
this.inputChannels = new LinkedHashSet<InputChannelBinding>(inputChannels);
}
public ChannelsMetadata getChannelsMetadata() {
ChannelsMetadata channels = new ChannelsMetadata();
channels.setModule(this.module);
channels.setInputChannels(new LinkedHashSet<InputChannelBinding>(this.inputChannels));
channels.setOutputChannels(new LinkedHashSet<OutputChannelBinding>(this.outputChannels));
return channels;
}
public OutputChannelBinding getOutputChannel(String name) {
if (name == null) {
return null;
}
for (OutputChannelBinding binding : this.outputChannels) {
if (name.equals(binding.getRemoteName())) {
return binding;
}
}
for (OutputChannelBinding binding : this.outputChannels) {
if (name.equals(binding.getLocalName())) {
return binding;
}
}
return null;
}
public InputChannelBinding getInputChannel(String name) {
if (name == null) {
return null;
}
for (InputChannelBinding binding : this.inputChannels) {
if (name.equals(binding.getRemoteName())) {
return binding;
}
}
for (InputChannelBinding binding : this.inputChannels) {
if (name.equals(binding.getLocalName())) {
return binding;
}
}
return null;
}
public void tap(String outputChannel) {
OutputChannelBinding channel = getOutputChannel(outputChannel);
if (channel == null || channel.isTapped()) {
return;
}
createAndBindTapChannel(channel.getTapChannelName(), channel.getLocalName());
channel.setTapped(true);
}
public void untap(String outputChannel) {
OutputChannelBinding channel = getOutputChannel(outputChannel);
if (channel == null || !channel.isTapped()) {
return;
}
String tapChannelName = channel.getTapChannelName();
this.binder.unbindProducers(tapChannelName);
channel.setTapped(false);
}
@ManagedOperation
public void rebind() {
boolean runnable = locateChannels();
if (runnable && !this.running) {
start();
}
if (!runnable && this.running) {
stop();
}
}
@Override
@ManagedOperation
public void start() {
if (!this.running) {
// Start everything, but don't call ourselves
if (!this.active.get()) {
if (this.active.compareAndSet(false, true)) {
boolean ready = bindChannels();
if (ready) {
this.running = true;
this.applicationContext.start();
}
this.active.set(false);
}
}
}
}
@Override
@ManagedOperation
public void stop() {
if (this.running) {
if (!this.active.get()) {
if (this.active.compareAndSet(false, true)) {
unbindChannels();
this.applicationContext.stop();
this.active.set(false);
}
}
}
this.running = false;
}
@Override
@ManagedAttribute
public boolean isRunning() {
return this.running && this.applicationContext.isRunning();
}
protected final void unbindChannels() {
for (InputChannelBinding binding : this.inputChannels) {
String name = this.bindings.get(binding.getRemoteName());
if (name == null) {
continue;
}
this.binder.unbindConsumers(name);
}
for (OutputChannelBinding binding : this.outputChannels) {
String name = this.bindings.get(binding.getRemoteName());
if (name == null) {
continue;
}
this.binder.unbindProducers(name);
if (binding.isTapped()) {
String tapChannelName = binding.getTapChannelName();
this.binder.unbindProducers(tapChannelName);
}
}
}
protected final boolean bindChannels() {
if (!locateChannels()) {
return false;
}
Map<String, Object> historyProperties = new LinkedHashMap<String, Object>();
if (this.trackHistory) {
// TODO: addHistoryTag();
}
for (OutputChannelBinding binding : this.outputChannels) {
String name = binding.getRemoteName();
MessageChannel outputChannel = this.channelResolver.resolveDestination(binding.getLocalName());
bindMessageProducer(outputChannel, name, this.module.getProducerProperties());
if (binding.isTapped()) {
String tapChannelName = this.channelLocator.tap(name);
binding.setTapChannelName(tapChannelName);
// tappableChannels.put(tapChannelName, outputChannel);
// if (isTapActive(tapChannelName)) {
createAndBindTapChannel(tapChannelName, name);
// }
}
if (this.trackHistory) {
historyProperties.put("outputChannel", name);
track(outputChannel, historyProperties);
}
}
for (InputChannelBinding binding : this.inputChannels) {
String name = binding.getRemoteName();
MessageChannel inputChannel = this.channelResolver.resolveDestination(binding.getLocalName());
bindMessageConsumer(inputChannel, name, this.module.getConsumerProperties());
if (this.trackHistory && this.outputChannels.size() != 1) {
historyProperties.put("inputChannel", name);
track(inputChannel, historyProperties);
}
}
return true;
}
private boolean locateChannels() {
logger.info("Locating channels");
boolean located = true;
for (OutputChannelBinding binding : this.outputChannels) {
String name = this.channelLocator.locate(binding.getLocalName());
if (name == null) {
logger.info("No channel found for: " + binding.getLocalName());
located = false;
}
binding.setRemoteName(name);
this.bindings.put(binding.getRemoteName(), name);
}
for (InputChannelBinding binding : this.inputChannels) {
String name = this.channelLocator.locate(binding.getLocalName());
if (name == null) {
logger.info("No channel found for: " + binding.getLocalName());
located = false;
}
binding.setRemoteName(name);
this.bindings.put(binding.getRemoteName(), name);
}
return located;
}
/*
* Following methods copied from parent to support the bindChannels() method above
*/
private void bindMessageConsumer(MessageChannel inputChannel,
String inputChannelName, Properties consumerProperties) {
if (isChannelPubSub(inputChannelName)) {
this.binder.bindPubSubConsumer(inputChannelName, inputChannel, consumerProperties);
}
else {
this.binder.bindConsumer(inputChannelName, inputChannel, consumerProperties);
}
}
private void bindMessageProducer(MessageChannel outputChannel,
String outputChannelName, Properties producerProperties) {
if (isChannelPubSub(outputChannelName)) {
this.binder.bindPubSubProducer(outputChannelName, outputChannel, producerProperties);
}
else {
this.binder.bindProducer(outputChannelName, outputChannel, producerProperties);
}
}
private boolean isChannelPubSub(String channelName) {
Assert.isTrue(StringUtils.hasText(channelName), "Channel name should not be empty/null.");
return (channelName.startsWith("tap:") || channelName.startsWith("topic:"));
}
/**
* Creates a wiretap on the output channel and binds the tap channel to
* {@link org.springframework.cloud.stream.binder.Binder}'s message target.
*
* @param tapChannelName the name of the tap channel
* @param localName the channel to tap
*/
private void createAndBindTapChannel(String tapChannelName, String localName) {
logger.info("creating and binding tap channel for {}", tapChannelName);
MessageChannel channel = this.channelResolver.resolveDestination(localName);
if (channel instanceof ChannelInterceptorAware) {
DirectChannel tapChannel = new DirectChannel();
tapChannel.setBeanName(tapChannelName + ".tap.bridge");
this.binder.bindPubSubProducer(tapChannelName, tapChannel, null); // TODO
// tap
// producer
// props
tapOutputChannel(tapChannel, (ChannelInterceptorAware) channel);
}
else {
if (logger.isDebugEnabled()) {
logger.debug("output channel is not interceptor aware. Tap will not be created.");
}
}
}
private MessageChannel tapOutputChannel(MessageChannel tapChannel, ChannelInterceptorAware outputChannel) {
outputChannel.addInterceptor(new WireTap(tapChannel));
return tapChannel;
}
private void track(MessageChannel channel, final Map<String, Object> historyProps) {
if (channel instanceof ChannelInterceptorAware) {
((ChannelInterceptorAware) channel)
.addInterceptor(new ChannelInterceptorAdapter() {
@Override
public Message<?> preSend(Message<?> message, MessageChannel channel) {
@SuppressWarnings("unchecked")
Collection<Map<String, Object>> history = (Collection<Map<String, Object>>) message
.getHeaders().get(BinderHeaders.BINDER_HISTORY);
if (history == null) {
history = new ArrayList<Map<String, Object>>(1);
}
else {
history = new ArrayList<Map<String, Object>>(history);
}
Map<String, Object> map = new LinkedHashMap<String, Object>();
map.putAll(historyProps);
map.put("thread", Thread.currentThread().getName());
history.add(map);
Message<?> out = ChannelBindingAdapter.this.messageBuilderFactory.fromMessage(message)
.setHeader(BinderHeaders.BINDER_HISTORY, history).build();
return out;
}
});
}
}
@Override
public boolean isAutoStartup() {
return true;
}
@Override
public void stop(Runnable callback) {
stop();
callback.run();
}
/**
* Return the lowest value to start this bean before any message producing lifecycle beans.
*/
@Override
public int getPhase() {
return Integer.MIN_VALUE;
}
}

57
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/adapter/ChannelsMetadata.java
View File

@@ -1,57 +0,0 @@
/*
* Copyright 2015 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.adapter;
import java.util.Collection;
import java.util.Collections;
import org.springframework.cloud.stream.config.ChannelBindingProperties;
/**
* @author Dave Syer
*/
public class ChannelsMetadata {
private Collection<OutputChannelBinding> outputChannels = Collections.emptySet();
private Collection<InputChannelBinding> inputChannels = Collections.emptySet();
private ChannelBindingProperties module;
public ChannelBindingProperties getModule() {
return this.module;
}
public void setModule(ChannelBindingProperties module) {
this.module = module;
}
public Collection<OutputChannelBinding> getOutputChannels() {
return this.outputChannels;
}
public void setOutputChannels(Collection<OutputChannelBinding> outputChannels) {
this.outputChannels = outputChannels;
}
public Collection<InputChannelBinding> getInputChannels() {
return this.inputChannels;
}
public void setInputChannels(Collection<InputChannelBinding> inputChannels) {
this.inputChannels = inputChannels;
}
}

51
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/adapter/DefaultChannelLocator.java
View File

@@ -1,51 +0,0 @@
/*
* Copyright 2015 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.adapter;
import org.springframework.cloud.stream.config.ChannelBindingProperties;
import org.springframework.util.StringUtils;
/**
* @author Dave Syer
*/
public class DefaultChannelLocator implements ChannelLocator {
private ChannelBindingProperties module;
public DefaultChannelLocator(ChannelBindingProperties module) {
this.module = module;
}
@Override
public String locate(String name) {
return module.getBindingPath(name);
}
@Override
public String tap(String name) {
return this.module.getTapChannelName(getPlainBinding(name));
}
private String getPlainBinding(String name) {
// remove prefixes such as 'tap:','topic:','queue:'
if (name.contains(":")) {
name = name.substring(name.indexOf(":") + 1);
}
return name;
}
}

52
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/adapter/OutputChannelBinding.java
View File

@@ -1,52 +0,0 @@
/*
* Copyright 2015 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.adapter;
/**
* @author Dave Syer
*/
public class OutputChannelBinding extends ChannelBinding {
private boolean tapped = false;
private String tapChannelName;
protected OutputChannelBinding() {
this(null);
}
public OutputChannelBinding(String localName) {
super(localName);
}
public boolean isTapped() {
return this.tapped;
}
public void setTapped(boolean tapped) {
this.tapped = tapped;
}
public void setTapChannelName(String tapChannelName) {
this.tapChannelName = tapChannelName;
}
public String getTapChannelName() {
return this.tapChannelName==null ? "" : this.tapChannelName;
}
}

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

@@ -0,0 +1,142 @@
/*
* Copyright 2015 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 org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.cloud.stream.annotation.EnableModule;
import org.springframework.cloud.stream.annotation.Processor;
import org.springframework.cloud.stream.annotation.Sink;
import org.springframework.cloud.stream.annotation.Source;
import org.springframework.cloud.stream.binding.BindableProxyFactory;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.integration.channel.DirectChannel;
/**
* @author Marius Bogoevici
*/
public class AggregateApplication {
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}
* modules by instantiating and binding them directly
*
* @param parentArgs arguments for the parent (prefixed with '--')
* @param modules a list module classes to be aggregated
* @param moduleArgs arguments for the modules (prefixed with '--")
*
* @return the resulting parent context for the aggregate
*/
public static ConfigurableApplicationContext run(Class<?>[] modules, String[] parentArgs, String[][] moduleArgs) {
ConfigurableApplicationContext parentContext = createParentContext(parentArgs != null ? parentArgs
: new String[0]);
runEmbedded(parentContext, modules, moduleArgs);
return parentContext;
}
public static ConfigurableApplicationContext run(Class<?>... modules) {
return run(modules, null, null);
}
/**
* Embeds a group of modules into an existing parent context
*
* @param parentContext the parent context
* @param modules a list of classes, representing root context definitions for modules
* @param args arguments for the modules
*/
public static void runEmbedded(ConfigurableApplicationContext parentContext,
Class<?>[] modules, String[][] args) {
SharedChannelRegistry bean = parentContext.getBean(SharedChannelRegistry.class);
prepareSharedChannelRegistry(bean, modules);
// create child contexts first
createChildContexts(parentContext, modules, args);
}
private static ConfigurableApplicationContext createParentContext(String[] args) {
SpringApplicationBuilder aggregatorParentConfiguration = new SpringApplicationBuilder();
aggregatorParentConfiguration
.sources(AggregatorParentConfiguration.class)
.web(false)
.headless(true)
.properties("spring.jmx.default-domain="
+ AggregatorParentConfiguration.class.getName());
return aggregatorParentConfiguration.run(args);
}
private static void createChildContexts(ConfigurableApplicationContext parentContext,
Class<?>[] modules, String args[][]) {
for (int i = modules.length - 1; i >= 0; i--) {
String moduleClassName = modules[i].getName();
embedModule(parentContext, getNamespace(moduleClassName, i), modules[i])
.run(args != null ? args[i] : new String[0]);
}
}
private static String getNamespace(String moduleClassName, int index) {
return moduleClassName + "_" + index;
}
private static SpringApplicationBuilder embedModule(
ConfigurableApplicationContext applicationContext, String namespace,
Class<?> module) {
return new SpringApplicationBuilder(module)
.web(false)
.showBanner(false)
.properties(BindableProxyFactory.CHANNEL_NAMESPACE_PROPERTY_NAME + "=" + namespace)
.registerShutdownHook(false)
.parent(applicationContext);
}
private static void prepareSharedChannelRegistry(SharedChannelRegistry sharedChannelRegistry, Class<?>[] modules) {
DirectChannel sharedChannel = null;
for (int i = 0; i < modules.length; i++) {
Class<?> module = modules[i];
String moduleClassName = module.getName();
if (i > 0) {
sharedChannelRegistry.register(getNamespace(moduleClassName, i)
+ "." + INPUT_CHANNEL_NAME, sharedChannel);
}
sharedChannel = new DirectChannel();
if (i < modules.length - 1) {
sharedChannelRegistry.register(getNamespace(moduleClassName, i)
+ "." + OUTPUT_CHANNEL_NAME, sharedChannel);
}
}
}
/**
* Basic configuration for a parent
*/
@EnableAutoConfiguration
@EnableModule
public static class AggregatorParentConfiguration {
@Bean
@ConditionalOnMissingBean(SharedChannelRegistry.class)
public SharedChannelRegistry sharedChannelRegistry() {
return new SharedChannelRegistry();
}
}
}

2
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/AggregateBuilder.java
View File

@@ -38,8 +38,6 @@ import org.springframework.util.StringUtils;
@ConfigurationProperties("spring.cloud.streams")
public class AggregateBuilder implements ApplicationContextAware {
public static final String DEFAULT_NAME = "application";
private ConfigurableApplicationContext parent;
private int index = 0;

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

@@ -0,0 +1,48 @@
/*
* Copyright 2015 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.Collections;
import java.util.Map;
import java.util.concurrent.ConcurrentSkipListMap;
import org.springframework.messaging.MessageChannel;
/**
* A registry for channels that can be shared between modules, used for module aggregation.
*
* @author Marius Bogoevici
*/
public class SharedChannelRegistry {
/**
* A {@link Map} of channels, indexed by name. A channel's name may be prefixed by a namespace.
*/
private Map<String, MessageChannel> sharedChannels = new ConcurrentSkipListMap<>(String.CASE_INSENSITIVE_ORDER);
public MessageChannel get(String id) {
return sharedChannels.get(id);
}
public void register(String id, MessageChannel messageChannel) {
sharedChannels.put(id, messageChannel);
}
public Map<String, MessageChannel> getAll() {
return Collections.unmodifiableMap(sharedChannels);
}
}

8
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/EnableModule.java
View File

@@ -24,10 +24,11 @@ import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
import org.springframework.cloud.stream.config.AggregateBuilderConfiguration;
import org.springframework.cloud.stream.config.ChannelBindingAdapterConfiguration;
import org.springframework.cloud.stream.config.ModuleRegistrar;
import org.springframework.cloud.stream.config.BindingBeansRegistrar;
import org.springframework.cloud.stream.config.ChannelBindingServiceConfiguration;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.integration.config.EnableIntegration;
/**
* Annotation that identifies a class as a module.
@@ -41,7 +42,8 @@ import org.springframework.context.annotation.Import;
@Documented
@Inherited
@Configuration
@Import({ChannelBindingAdapterConfiguration.class, AggregateBuilderConfiguration.class, ModuleRegistrar.class})
@Import({ChannelBindingServiceConfiguration.class, AggregateBuilderConfiguration.class, BindingBeansRegistrar.class})
@EnableIntegration
public @interface EnableModule {
Class<?>[] value() default {};

48
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/Bindable.java Normal file
View File

@@ -0,0 +1,48 @@
/*
* Copyright 2015 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.binding;
/**
* Marker interface for instances that can bind/unbind groups of inputs and outputs.
*
* Intended for internal use.
*
* @author Marius Bogoevici
*/
public interface Bindable {
/**
* Binds all the inputs associated with this instance.
*/
void bindInputs(ChannelBindingService adapter);
/**
* Binds all the outputs associated with this instance.
*/
void bindOutputs(ChannelBindingService adapter);
/**
* Unbinds all the inputs associated with this instance.
*/
void unbindInputs(ChannelBindingService adapter);
/**
* Unbinds all the outputs associated with this instance.
*/
void unbindOutputs(ChannelBindingService adapter);
}

348
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindableProxyFactory.java Normal file
View File

@@ -0,0 +1,348 @@
/*
* Copyright 2015 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.binding;
import java.lang.reflect.Method;
import java.util.HashMap;
import java.util.Map;
import org.aopalliance.intercept.MethodInterceptor;
import org.aopalliance.intercept.MethodInvocation;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.aop.framework.ProxyFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.cloud.stream.aggregate.SharedChannelRegistry;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.binder.MessageChannelBinderSupport;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.config.ConsumerEndpointFactoryBean;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.scheduling.support.PeriodicTrigger;
import org.springframework.util.Assert;
import org.springframework.util.ReflectionUtils;
/**
* {@link FactoryBean} for instantiating the interfaces specified via
* {@link org.springframework.cloud.stream.annotation.EnableModule}
*
* @author Marius Bogoevici
* @author David Syer
*
* @see org.springframework.cloud.stream.annotation.EnableModule
*/
public class BindableProxyFactory implements MethodInterceptor, FactoryBean<Object>,
BeanFactoryAware, Bindable, InitializingBean {
private static Log log = LogFactory.getLog(BindableProxyFactory.class);
public 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";
public static final String POLLABLE_BRIDGE_INTERVAL_PROPERTY_NAME = SPRING_CLOUD_STREAM_INTERNAL_PREFIX + ".pollableBridge.interval";
private Class<?> type;
@Value("${" + CHANNEL_NAMESPACE_PROPERTY_NAME + ":}")
private String channelNamespace;
@Value("${" + POLLABLE_BRIDGE_INTERVAL_PROPERTY_NAME + ":1000}")
private int pollableBridgeDefaultFrequency;
private Object proxy = null;
private Map<String, ChannelHolder> inputs = new HashMap<>();
private Map<String, ChannelHolder> outputs = new HashMap<>();
private ConfigurableListableBeanFactory beanFactory;
@Autowired(required = false)
private SharedChannelRegistry sharedChannelRegistry;
public BindableProxyFactory(Class<?> type) {
this.type = type;
}
@Override
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
this.beanFactory = (ConfigurableListableBeanFactory) beanFactory;
}
@Override
public void afterPropertiesSet() throws Exception {
Assert.notNull(beanFactory, "Bean factory cannot be empty");
}
private void createChannels(Class<?> type) throws Exception {
ReflectionUtils.doWithMethods(type, new ReflectionUtils.MethodCallback() {
@Override
public void doWith(Method method) throws IllegalArgumentException,
IllegalAccessException {
Input input = AnnotationUtils.findAnnotation(method, Input.class);
if (input != null) {
String name = BindingBeanDefinitionRegistryUtils.getChannelName(
input, method);
Class<?> inputChannelType = method.getReturnType();
MessageChannel sharedChannel = locateSharedChannel(name);
if (sharedChannel == null) {
MessageChannel inputChannel = createMessageChannel(inputChannelType);
inputs.put(name, new ChannelHolder(inputChannel, true));
}
else {
if (inputChannelType.isAssignableFrom(sharedChannel.getClass())) {
inputs.put(name, new ChannelHolder(sharedChannel, false));
}
else {
// handle the special case where the shared channel is of a different nature
// (i.e. pollable vs subscribable) than the target channel
final MessageChannel inputChannel = createMessageChannel(inputChannelType);
if (isPollable(sharedChannel.getClass())) {
bridgePollableToSubscribableChannel(sharedChannel,
inputChannel);
}
else {
bridgeSubscribableToPollableChannel(
(SubscribableChannel) sharedChannel, inputChannel);
}
inputs.put(name, new ChannelHolder(inputChannel, false));
}
}
}
Output output = AnnotationUtils.findAnnotation(method, Output.class);
if (output != null) {
String name = BindingBeanDefinitionRegistryUtils.getChannelName(
output, method);
Class<?> messageChannelType = method.getReturnType();
MessageChannel sharedChannel = locateSharedChannel(name);
if (sharedChannel == null) {
MessageChannel outputChannel = createMessageChannel(messageChannelType);
outputs.put(name, new ChannelHolder(outputChannel, true));
}
else {
if (messageChannelType.isAssignableFrom(sharedChannel.getClass())) {
outputs.put(name, new ChannelHolder(sharedChannel, false));
}
else {
// handle the special case where the shared channel is of a different nature
// (i.e. pollable vs subscribable) than the target channel
final MessageChannel outputChannel = createMessageChannel(messageChannelType);
if (isPollable(messageChannelType)) {
bridgePollableToSubscribableChannel(outputChannel,
sharedChannel);
}
else {
bridgeSubscribableToPollableChannel(
(SubscribableChannel) outputChannel,
sharedChannel);
}
outputs.put(name, new ChannelHolder(outputChannel, false));
}
}
}
}
});
}
private MessageChannel locateSharedChannel(String name) {
return sharedChannelRegistry != null ? sharedChannelRegistry.get(getNamespacePrefixedChannelName(name)) : null;
}
private String getNamespacePrefixedChannelName(String name) {
return channelNamespace + "." + name;
}
private void bridgeSubscribableToPollableChannel(SubscribableChannel sharedChannel,
MessageChannel inputChannel) {
sharedChannel.subscribe(new MessageChannelBinderSupport.DirectHandler(
inputChannel));
}
private void bridgePollableToSubscribableChannel(MessageChannel pollableChannel,
MessageChannel subscribableChannel) {
ConsumerEndpointFactoryBean consumerEndpointFactoryBean = new ConsumerEndpointFactoryBean();
consumerEndpointFactoryBean.setInputChannel(pollableChannel);
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setTrigger(new PeriodicTrigger(pollableBridgeDefaultFrequency));
consumerEndpointFactoryBean.setPollerMetadata(pollerMetadata);
consumerEndpointFactoryBean
.setHandler(new MessageChannelBinderSupport.DirectHandler(
subscribableChannel));
consumerEndpointFactoryBean.setBeanFactory(beanFactory);
try {
consumerEndpointFactoryBean.afterPropertiesSet();
} catch (Exception e) {
throw new IllegalStateException(e);
}
consumerEndpointFactoryBean.start();
}
private MessageChannel createMessageChannel(Class<?> messageChannelType) {
return isPollable(messageChannelType) ? new QueueChannel() : new DirectChannel();
}
private boolean isPollable(Class<?> channelType) {
return PollableChannel.class.equals(channelType);
}
@Override
public synchronized Object invoke(MethodInvocation invocation) throws Throwable {
Method method = invocation.getMethod();
if (MessageChannel.class.isAssignableFrom(method.getReturnType())) {
Input input = AnnotationUtils.findAnnotation(method, Input.class);
if (input != null) {
String name = BindingBeanDefinitionRegistryUtils.getChannelName(input,
method);
return this.inputs.get(name).getMessageChannel();
}
Output output = AnnotationUtils.findAnnotation(method, Output.class);
if (output != null) {
String name = BindingBeanDefinitionRegistryUtils.getChannelName(output,
method);
return this.outputs.get(name).getMessageChannel();
}
}
// ignore
return null;
}
@Override
public synchronized Object getObject() throws Exception {
if (this.proxy == null) {
createChannels(this.type);
ProxyFactory factory = new ProxyFactory(type, this);
this.proxy = factory.getProxy();
}
return this.proxy;
}
@Override
public Class<?> getObjectType() {
return this.type;
}
@Override
public boolean isSingleton() {
return true;
}
@Override
public void bindInputs(ChannelBindingService channelBindingService) {
if (log.isDebugEnabled()) {
log.debug(String.format("Binding inputs for %s:%s", this.channelNamespace, this.type));
}
for (Map.Entry<String, ChannelHolder> channelHolderEntry : inputs.entrySet()) {
ChannelHolder channelHolder = channelHolderEntry.getValue();
if (channelHolder.isBindable()) {
if (log.isDebugEnabled()) {
log.debug(String.format("Binding %s:%s:%s", this.channelNamespace, this.type, channelHolderEntry.getKey()));
}
channelBindingService.bindConsumer(
channelHolder.getMessageChannel(), channelHolderEntry.getKey());
}
}
}
@Override
public void bindOutputs(ChannelBindingService channelBindingService) {
if (log.isDebugEnabled()) {
log.debug(String.format("Binding outputs for %s:%s", this.channelNamespace, this.type));
}
for (Map.Entry<String, ChannelHolder> channelHolderEntry : outputs.entrySet()) {
if (channelHolderEntry.getValue().isBindable()) {
if (log.isDebugEnabled()) {
log.debug(String.format("Binding %s:%s:%s", this.channelNamespace, this.type, channelHolderEntry.getKey()));
}
channelBindingService.bindProducer(channelHolderEntry.getValue()
.getMessageChannel(), channelHolderEntry.getKey());
}
}
}
@Override
public void unbindInputs(ChannelBindingService channelBindingService) {
if (log.isDebugEnabled()) {
log.debug(String.format("Unbinding inputs for %s:%s", this.channelNamespace, this.type));
}
for (Map.Entry<String, ChannelHolder> channelHolderEntry : inputs.entrySet()) {
if (channelHolderEntry.getValue().isBindable()) {
if (log.isDebugEnabled()) {
log.debug(String.format("Unbinding %s:%s:%s", this.channelNamespace, this.type, channelHolderEntry.getKey()));
}
channelBindingService.unbindConsumers(channelHolderEntry.getKey());
}
}
}
@Override
public void unbindOutputs(ChannelBindingService channelBindingService) {
if (log.isDebugEnabled()) {
log.debug(String.format("Unbinding outputs for %s:%s", this.channelNamespace, this.type));
}
for (Map.Entry<String, ChannelHolder> channelHolderEntry : outputs.entrySet()) {
if (channelHolderEntry.getValue().isBindable()) {
if (log.isDebugEnabled()) {
log.debug(String.format("Binding %s:%s:%s", this.channelNamespace, this.type, channelHolderEntry.getKey()));
}
channelBindingService.unbindProducers(channelHolderEntry.getKey());
}
}
}
/**
* Holds information about the channels exposed by the interface proxy, as well as
* their status.
*
*/
static class ChannelHolder {
private MessageChannel messageChannel;
private boolean bindable;
public ChannelHolder(MessageChannel messageChannel, boolean bindable) {
this.messageChannel = messageChannel;
this.bindable = bindable;
}
public MessageChannel getMessageChannel() {
return messageChannel;
}
public boolean isBindable() {
return bindable;
}
}
}

3
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderAwareChannelResolver.java → spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BinderAwareChannelResolver.java
View File

@@ -14,10 +14,11 @@
* limitations under the License.
*/
package org.springframework.cloud.stream.binder;
package org.springframework.cloud.stream.binding;
import java.util.Properties;
import org.springframework.cloud.stream.binder.Binder;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.core.BeanFactoryMessageChannelDestinationResolver;
import org.springframework.messaging.core.DestinationResolutionException;

2
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderAwareRouterBeanPostProcessor.java → spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BinderAwareRouterBeanPostProcessor.java
View File

@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.cloud.stream.binder;
package org.springframework.cloud.stream.binding;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.config.BeanPostProcessor;

125
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingBeanDefinitionRegistryUtils.java Normal file
View File

@@ -0,0 +1,125 @@
/*
* Copyright 2015 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.binding;
import java.lang.annotation.Annotation;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import org.springframework.beans.factory.support.AutowireCandidateQualifier;
import org.springframework.beans.factory.support.BeanDefinitionRegistry;
import org.springframework.beans.factory.support.RootBeanDefinition;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.ModuleChannels;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.util.ReflectionUtils;
import org.springframework.util.ReflectionUtils.MethodCallback;
import org.springframework.util.StringUtils;
/**
* Utility class for registering bean definitions for message channels.
*
* @author Marius Bogoevici
* @author Dave Syer
*/
public abstract class BindingBeanDefinitionRegistryUtils {
public static void registerInputChannelBeanDefinition(String qualifierValue,
String name, String channelInterfaceBeanName,
String channelInterfaceMethodName, BeanDefinitionRegistry registry) {
registerChannelBeanDefinition(Input.class, qualifierValue, name,
channelInterfaceBeanName, channelInterfaceMethodName, registry);
}
public static void registerOutputChannelBeanDefinition(String qualifierValue,
String name, String channelInterfaceBeanName,
String channelInterfaceMethodName, BeanDefinitionRegistry registry) {
registerChannelBeanDefinition(Output.class, qualifierValue, name,
channelInterfaceBeanName, channelInterfaceMethodName, registry);
}
private static void registerChannelBeanDefinition(
Class<? extends Annotation> qualifier, String qualifierValue, String name,
String channelInterfaceBeanName, String channelInterfaceMethodName,
BeanDefinitionRegistry registry) {
RootBeanDefinition rootBeanDefinition = new RootBeanDefinition();
rootBeanDefinition.setFactoryBeanName(channelInterfaceBeanName);
rootBeanDefinition.setUniqueFactoryMethodName(channelInterfaceMethodName);
rootBeanDefinition.addQualifier(new AutowireCandidateQualifier(qualifier,
qualifierValue));
registry.registerBeanDefinition(name, rootBeanDefinition);
}
public static void registerChannelBeanDefinitions(Class<?> type,
final String channelInterfaceBeanName, final BeanDefinitionRegistry registry) {
final List<String> channelNames = new ArrayList<>();
ReflectionUtils.doWithMethods(type, new MethodCallback() {
@Override
public void doWith(Method method) throws IllegalArgumentException,
IllegalAccessException {
Input input = AnnotationUtils.findAnnotation(method, Input.class);
if (input != null) {
String name = getChannelName(input, method);
registerInputChannelBeanDefinition(input.value(), name,
channelInterfaceBeanName, method.getName(), registry);
}
Output output = AnnotationUtils.findAnnotation(method, Output.class);
if (output != null) {
String name = getChannelName(output, method);
registerOutputChannelBeanDefinition(output.value(), name,
channelInterfaceBeanName, method.getName(), registry);
}
}
});
}
public static void registerChannelsQualifiedBeanDefinitions(Class<?> parent,
Class<?> type, final BeanDefinitionRegistry registry) {
if (type.isInterface()) {
RootBeanDefinition rootBeanDefinition = new RootBeanDefinition(
BindableProxyFactory.class);
rootBeanDefinition.addQualifier(new AutowireCandidateQualifier(
ModuleChannels.class, parent));
rootBeanDefinition.getConstructorArgumentValues().addGenericArgumentValue(
type);
registry.registerBeanDefinition(type.getName(), rootBeanDefinition);
}
else {
RootBeanDefinition rootBeanDefinition = new RootBeanDefinition(type);
rootBeanDefinition.addQualifier(new AutowireCandidateQualifier(
ModuleChannels.class, parent));
registry.registerBeanDefinition(type.getName(), rootBeanDefinition);
}
}
public static String getChannelName(Annotation annotation, Method method) {
Map<String, Object> attrs = AnnotationUtils.getAnnotationAttributes(annotation,
false);
if (attrs.containsKey("value")
&& StringUtils.hasText((CharSequence) attrs.get("value"))) {
return (String) attrs.get("value");
}
return method.getName();
}
}

121
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ChannelBindingLifecycle.java Normal file
View File

@@ -0,0 +1,121 @@
/*
* Copyright 2015 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.binding;
import java.util.Map;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.SmartLifecycle;
/**
* Coordinates binding/unbinding of input/output channels in accordance to the lifecycle of the host context.
*
* @author Marius Bogoevici
* @author Ilayaperumal Gopinathan
*/
public class ChannelBindingLifecycle implements SmartLifecycle, ApplicationContextAware {
private volatile boolean running = false;
private final Object lifecycleMonitor = new Object();
private ConfigurableApplicationContext applicationContext;
@Override
public void setApplicationContext(ApplicationContext applicationContext)
throws BeansException {
this.applicationContext = (ConfigurableApplicationContext) applicationContext;
}
@Override
public void start() {
if (!running) {
synchronized (lifecycleMonitor) {
if (!running) {
// retrieve the ChannelBindingService lazily, avoiding early initialization
try {
ChannelBindingService channelBindingService = this.applicationContext.getBean(ChannelBindingService.class);
Map<String, Bindable> bindables = this.applicationContext.getBeansOfType(Bindable.class);
for (Bindable bindable : bindables.values()) {
bindable.bindOutputs(channelBindingService);
}
for (Bindable bindable : bindables.values()) {
bindable.bindInputs(channelBindingService);
}
} catch (BeansException e) {
throw new IllegalStateException("Cannot perform binding, no proper implementation found",e);
}
this.running = true;
}
}
}
}
@Override
public void stop() {
if (running) {
synchronized (lifecycleMonitor) {
if (running) {
try {
// retrieve the ChannelBindingService lazily, avoiding early initialization
ChannelBindingService channelBindingService = this.applicationContext.getBean(ChannelBindingService.class);
Map<String, Bindable> bindables = this.applicationContext.getBeansOfType(Bindable.class);
for (Bindable bindable : bindables.values()) {
bindable.unbindInputs(channelBindingService);
}
for (Bindable bindable : bindables.values()) {
bindable.unbindOutputs(channelBindingService);
}
} catch (BeansException e) {
throw new IllegalStateException("Cannot perform binding, no proper implementation found",e);
}
this.running = false;
}
}
}
}
@Override
public boolean isRunning() {
return running;
}
@Override
public boolean isAutoStartup() {
return true;
}
@Override
public void stop(Runnable callback) {
stop();
if (callback != null) {
callback.run();
}
}
/**
* Return the lowest value to start this bean before any message producing lifecycle
* beans.
*/
@Override
public int getPhase() {
return Integer.MIN_VALUE;
}
}

87
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ChannelBindingService.java Normal file
View File

@@ -0,0 +1,87 @@
/*
* Copyright 2015 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.binding;
import org.springframework.cloud.stream.binder.Binder;
import org.springframework.cloud.stream.config.ChannelBindingProperties;
import org.springframework.messaging.MessageChannel;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
/**
* Handles the binding of input/output channels by delegating to an underlying {@link Binder}.
*
* @author Mark Fisher
* @author Dave Syer
* @author Marius Bogoevici
*/
public class ChannelBindingService {
private Binder<MessageChannel> binder;
private ChannelBindingProperties channelBindingProperties;
public ChannelBindingService(ChannelBindingProperties channelBindingProperties,
Binder<MessageChannel> binder) {
this.channelBindingProperties = channelBindingProperties;
this.binder = binder;
}
public void bindConsumer(MessageChannel inputChannel, String inputChannelName) {
String channelBindingTarget = this.channelBindingProperties
.getBindingPath(inputChannelName);
if (isChannelPubSub(inputChannelName)) {
this.binder.bindPubSubConsumer(channelBindingTarget, inputChannel,
this.channelBindingProperties.getConsumerProperties());
}
else {
this.binder.bindConsumer(channelBindingTarget, inputChannel,
this.channelBindingProperties.getConsumerProperties());
}
}
public void bindProducer(MessageChannel outputChannel, String outputChannelName) {
String channelBindingTarget = this.channelBindingProperties
.getBindingPath(outputChannelName);
if (isChannelPubSub(outputChannelName)) {
this.binder.bindPubSubProducer(channelBindingTarget, outputChannel,
this.channelBindingProperties.getProducerProperties());
}
else {
this.binder.bindProducer(channelBindingTarget, outputChannel,
this.channelBindingProperties.getProducerProperties());
}
}
private boolean isChannelPubSub(String channelName) {
Assert.isTrue(StringUtils.hasText(channelName),
"Channel name should not be empty/null.");
return channelName.startsWith("topic:");
}
public void unbindConsumers(String inputChannelName) {
this.binder.unbindConsumers
(inputChannelName);
}
public void unbindProducers(String outputChannelName) {
this.binder.unbindProducers(outputChannelName);
}
}

28
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ModuleRegistrar.java → spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingBeansRegistrar.java
View File

@@ -18,16 +18,11 @@ package org.springframework.cloud.stream.config;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import org.springframework.beans.factory.support.BeanDefinitionRegistry;
import org.springframework.cloud.stream.annotation.EnableModule;
import org.springframework.cloud.stream.utils.MessageChannelBeanDefinitionRegistryUtils;
import org.springframework.context.EnvironmentAware;
import org.springframework.cloud.stream.binding.BindingBeanDefinitionRegistryUtils;
import org.springframework.context.annotation.ImportBeanDefinitionRegistrar;
import org.springframework.core.env.ConfigurableEnvironment;
import org.springframework.core.env.Environment;
import org.springframework.core.env.PropertiesPropertySource;
import org.springframework.core.type.AnnotationMetadata;
import org.springframework.util.ClassUtils;
import org.springframework.util.MultiValueMap;
@@ -36,16 +31,8 @@ import org.springframework.util.MultiValueMap;
* @author Marius Bogoevici
* @author Dave Syer
*/
public class ModuleRegistrar implements ImportBeanDefinitionRegistrar, EnvironmentAware {
public static final String SPRING_CLOUD_STREAM_BINDINGS_PREFIX = "spring.cloud.stream.bindings";
private ConfigurableEnvironment environment;
@Override
public void setEnvironment(Environment environment) {
this.environment = (ConfigurableEnvironment) environment;
}
public class BindingBeansRegistrar implements ImportBeanDefinitionRegistrar {
@Override
public void registerBeanDefinitions(AnnotationMetadata metadata,
@@ -54,18 +41,11 @@ public class ModuleRegistrar implements ImportBeanDefinitionRegistrar, Environme
EnableModule.class.getName(), false);
List<String> registeredChannelNames = new ArrayList<>();
for (Class<?> type : collectClasses(attributes.get("value"))) {
registeredChannelNames.addAll(MessageChannelBeanDefinitionRegistryUtils.registerChannelBeanDefinitions(type, registry));
MessageChannelBeanDefinitionRegistryUtils.registerChannelsQualifiedBeanDefinitions(
BindingBeanDefinitionRegistryUtils.registerChannelBeanDefinitions(type, type.getName(), registry);
BindingBeanDefinitionRegistryUtils.registerChannelsQualifiedBeanDefinitions(
ClassUtils.resolveClassName(metadata.getClassName(), null), type,
registry);
}
Properties defaultChannelNameProperties = new Properties();
for (String registeredChannelName : registeredChannelNames) {
defaultChannelNameProperties.put(SPRING_CLOUD_STREAM_BINDINGS_PREFIX + "." + registeredChannelName,
"${spring.application.name:spring.cloud.stream}" + "." + registeredChannelName);
}
environment.getPropertySources().addLast(
new PropertiesPropertySource("default-spring-cloud-stream-channel-bindings", defaultChannelNameProperties));
}
private List<Class<?>> collectClasses(List<Object> list) {

146
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingAdapterConfiguration.java
View File

@@ -1,146 +0,0 @@
/*
* Copyright 2015 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.config;
import java.util.Collection;
import java.util.LinkedHashSet;
import java.util.Properties;
import java.util.Set;
import org.springframework.beans.factory.BeanFactoryUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.cloud.stream.adapter.ChannelBindingAdapter;
import org.springframework.cloud.stream.adapter.DefaultChannelLocator;
import org.springframework.cloud.stream.adapter.InputChannelBinding;
import org.springframework.cloud.stream.adapter.OutputChannelBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.binder.Binder;
import org.springframework.cloud.stream.binder.BinderAwareChannelResolver;
import org.springframework.cloud.stream.binder.BinderAwareRouterBeanPostProcessor;
import org.springframework.cloud.stream.endpoint.ChannelsEndpoint;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.core.DestinationResolutionException;
import org.springframework.messaging.core.DestinationResolver;
/**
* Configuration class that provides necessary beans for {@link MessageChannel} binding.
*
* @author Dave Syer
* @author David Turanski
* @author Marius Bogoevici
* @author Ilayaperumal Gopinathan
*/
@Configuration
@EnableConfigurationProperties(ChannelBindingProperties.class)
public class ChannelBindingAdapterConfiguration {
@Autowired
private ChannelBindingProperties module;
@Autowired
private ConfigurableListableBeanFactory beanFactory;
@Autowired
private Binder<MessageChannel> binder;
@Bean
public ChannelBindingAdapter bindingAdapter() {
ChannelBindingAdapter adapter = new ChannelBindingAdapter(this.module,
this.binder);
adapter.setChannelLocator(new DefaultChannelLocator(this.module));
adapter.setOutputChannels(getOutputChannels());
adapter.setInputChannels(getInputChannels());
adapter.setChannelResolver(binderAwareChannelResolver());
return adapter;
}
@Bean
public ChannelsEndpoint channelsEndpoint(ChannelBindingAdapter adapter) {
return new ChannelsEndpoint(adapter);
}
Collection<OutputChannelBinding> getOutputChannels() {
Set<OutputChannelBinding> channels = new LinkedHashSet<>();
String[] names = this.beanFactory.getBeanNamesForType(MessageChannel.class);
for (String name : names) {
BeanDefinition beanDefinition = this.beanFactory.getBeanDefinition(name);
// for now, just assume that the beans are at least AbstractBeanDefinition
if (beanDefinition instanceof AbstractBeanDefinition
&& ((AbstractBeanDefinition) beanDefinition)
.getQualifier(Output.class.getName()) != null) {
channels.add(new OutputChannelBinding(name));
}
}
return channels;
}
Collection<InputChannelBinding> getInputChannels() {
Set<InputChannelBinding> channels = new LinkedHashSet<>();
String[] names = this.beanFactory.getBeanNamesForType(MessageChannel.class);
for (String name : names) {
BeanDefinition beanDefinition = this.beanFactory.getBeanDefinition(name);
// for now, just assume that the beans are at least AbstractBeanDefinition
if (beanDefinition instanceof AbstractBeanDefinition
&& ((AbstractBeanDefinition) beanDefinition).getQualifier(Input.class
.getName()) != null) {
channels.add(new InputChannelBinding(name));
}
}
return channels;
}
@Bean
public BinderAwareChannelResolver binderAwareChannelResolver() {
return new BinderAwareChannelResolver(this.binder, new Properties());
}
// IMPORTANT: Nested class to avoid instantiating all of the above early
@Configuration
protected static class PostProcessorConfiguration {
private BinderAwareChannelResolver binderAwareChannelResolver;
@Bean
public BinderAwareRouterBeanPostProcessor binderAwareRouterBeanPostProcessor(
final ConfigurableListableBeanFactory beanFactory) {
// IMPORTANT: Lazy delegate to avoid instantiating all of the above early
return new BinderAwareRouterBeanPostProcessor(
new DestinationResolver<MessageChannel>() {
@Override
public MessageChannel resolveDestination(String name)
throws DestinationResolutionException {
if (PostProcessorConfiguration.this.binderAwareChannelResolver == null) {
PostProcessorConfiguration.this.binderAwareChannelResolver = BeanFactoryUtils.beanOfType(
beanFactory, BinderAwareChannelResolver.class);
}
return PostProcessorConfiguration.this.binderAwareChannelResolver.resolveDestination(name);
}
});
}
}
}

2
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingProperties.java
View File

@@ -87,5 +87,5 @@ public class ChannelBindingProperties {
public String getTapChannelName(String channelName) {
return "tap:" + getBindingPath(channelName);
}
}

97
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceConfiguration.java Normal file
View File

@@ -0,0 +1,97 @@
/*
* Copyright 2015 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.config;
import java.util.Properties;
import org.springframework.beans.factory.BeanFactoryUtils;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.cloud.stream.binding.ChannelBindingService;
import org.springframework.cloud.stream.binder.Binder;
import org.springframework.cloud.stream.binding.BinderAwareChannelResolver;
import org.springframework.cloud.stream.binding.BinderAwareRouterBeanPostProcessor;
import org.springframework.cloud.stream.binding.ChannelBindingLifecycle;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.DependsOn;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.core.DestinationResolutionException;
import org.springframework.messaging.core.DestinationResolver;
/**
* Configuration class that provides necessary beans for {@link MessageChannel} binding.
*
* @author Dave Syer
* @author David Turanski
* @author Marius Bogoevici
* @author Ilayaperumal Gopinathan
*/
@Configuration
@EnableConfigurationProperties(ChannelBindingProperties.class)
public class ChannelBindingServiceConfiguration {
@Bean
@ConditionalOnMissingBean(ChannelBindingService.class)
public ChannelBindingService bindingService(ChannelBindingProperties channelBindingProperties, Binder<MessageChannel> binder) {
return new ChannelBindingService(channelBindingProperties, binder);
}
@Bean
@DependsOn("bindingService")
public ChannelBindingLifecycle channelBindingLifecycle() {
return new ChannelBindingLifecycle();
}
@Bean
public BinderAwareChannelResolver binderAwareChannelResolver(
Binder<MessageChannel> binder) {
return new BinderAwareChannelResolver(binder, new Properties());
}
// IMPORTANT: Nested class to avoid instantiating all of the above early
@Configuration
protected static class PostProcessorConfiguration {
private BinderAwareChannelResolver binderAwareChannelResolver;
@Bean
public BinderAwareRouterBeanPostProcessor binderAwareRouterBeanPostProcessor(
final ConfigurableListableBeanFactory beanFactory) {
// IMPORTANT: Lazy delegate to avoid instantiating all of the above early
return new BinderAwareRouterBeanPostProcessor(
new DestinationResolver<MessageChannel>() {
@Override
public MessageChannel resolveDestination(String name)
throws DestinationResolutionException {
if (PostProcessorConfiguration.this.binderAwareChannelResolver == null) {
PostProcessorConfiguration.this.binderAwareChannelResolver = BeanFactoryUtils
.beanOfType(beanFactory,
BinderAwareChannelResolver.class);
}
return PostProcessorConfiguration.this.binderAwareChannelResolver
.resolveDestination(name);
}
});
}
}
}

83
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/DirectChannelFactoryBean.java
View File

@@ -1,83 +0,0 @@
/*
* Copyright 2015 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.config;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.util.Assert;
/**
* {@link FactoryBean} for creating channels for fields annotated with
* {@link org.springframework.cloud.stream.annotation.Input} and
* {@link org.springframework.cloud.stream.annotation.Output}.
*
* @author Marius Bogoevici
*/
public class DirectChannelFactoryBean implements FactoryBean<DirectChannel>, BeanNameAware, BeanFactoryAware, ApplicationContextAware {
private DirectChannel directChannel;
private String beanName;
private BeanFactory beanFactory;
private ApplicationContext applicationContext;
@Override
public void setBeanName(String beanName) {
this.beanName = beanName;
}
@Override
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
Assert.notNull(beanFactory, "'beanFactory' must not be null");
this.beanFactory = beanFactory;
}
@Override
public void setApplicationContext(ApplicationContext applicationContext) {
Assert.notNull(beanFactory, "'applicationContext' must not be null");
this.applicationContext = applicationContext;
}
@Override
public synchronized DirectChannel getObject() throws Exception {
if (directChannel == null) {
directChannel = new DirectChannel();
}
directChannel.setBeanName(beanName);
directChannel.setBeanFactory(beanFactory);
directChannel.setApplicationContext(applicationContext);
return directChannel;
}
@Override
public Class<?> getObjectType() {
return DirectChannel.class;
}
@Override
public boolean isSingleton() {
return true;
}
}

79
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/endpoint/ChannelsEndpoint.java
View File

@@ -1,79 +0,0 @@
/*
* Copyright 2015 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.endpoint;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import org.springframework.boot.actuate.endpoint.AbstractEndpoint;
import org.springframework.cloud.stream.adapter.ChannelsMetadata;
import org.springframework.cloud.stream.adapter.ChannelBindingAdapter;
import org.springframework.cloud.stream.adapter.OutputChannelBinding;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
/**
* @author Dave Syer
*/
@RestController
public class ChannelsEndpoint extends AbstractEndpoint<Map<String, ?>> {
private ChannelBindingAdapter adapter;
public ChannelsEndpoint(ChannelBindingAdapter adapter) {
super("channels");
this.adapter = adapter;
}
@RequestMapping(value="/channels/taps")
public List<OutputChannelBinding> taps() {
List<OutputChannelBinding> list = new ArrayList<OutputChannelBinding>();
for (OutputChannelBinding binding : adapter.getChannelsMetadata().getOutputChannels()) {
if (binding.isTapped()) {
list.add(binding);
}
}
return list ;
}
@RequestMapping(value="/channels/taps", method=RequestMethod.POST)
public OutputChannelBinding tap(@RequestParam String channel) {
adapter.tap(channel);
return adapter.getOutputChannel(channel);
}
@RequestMapping(value="/channels/taps", method=RequestMethod.DELETE)
public OutputChannelBinding untap(@RequestParam String channel) {
adapter.untap(channel);
return adapter.getOutputChannel(channel);
}
@Override
public Map<String, ?> invoke() {
LinkedHashMap<String, Object> map = new LinkedHashMap<String, Object>();
ChannelsMetadata channels = adapter.getChannelsMetadata();
map.put("inputChannels", channels.getInputChannels());
map.put("outputChannels", channels.getOutputChannels());
map.put("module", channels.getModule());
return map;
}
}

203
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/utils/MessageChannelBeanDefinitionRegistryUtils.java
View File

@@ -1,203 +0,0 @@
/*
* Copyright 2015 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.utils;
import java.lang.annotation.Annotation;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import org.aopalliance.intercept.MethodInterceptor;
import org.aopalliance.intercept.MethodInvocation;
import org.springframework.aop.framework.ProxyFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.beans.factory.support.AutowireCandidateQualifier;
import org.springframework.beans.factory.support.BeanDefinitionRegistry;
import org.springframework.beans.factory.support.RootBeanDefinition;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.ModuleChannels;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.config.DirectChannelFactoryBean;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.messaging.MessageChannel;
import org.springframework.util.ReflectionUtils;
import org.springframework.util.ReflectionUtils.MethodCallback;
import org.springframework.util.StringUtils;
/**
* Utility class for registering bean definitions for message channels.
*
* @author Marius Bogoevici
* @author Dave Syer
*/
public abstract class MessageChannelBeanDefinitionRegistryUtils {
public static final String DEFAULT_INPUT_QUALIFIER_VALUE;
public static final String DEFAULT_OUTPUT_QUALIFIER_VALUE;
static {
DEFAULT_INPUT_QUALIFIER_VALUE = (String) ReflectionUtils.findMethod(Input.class, "value").getDefaultValue();
DEFAULT_OUTPUT_QUALIFIER_VALUE = (String) ReflectionUtils.findMethod(Output.class, "value").getDefaultValue();
}
public static void registerInputChannelBeanDefinition(String name,
BeanDefinitionRegistry registry) {
registerInputChannelBeanDefinition(DEFAULT_INPUT_QUALIFIER_VALUE, name, registry);
}
public static void registerInputChannelBeanDefinition(String qualifierValue, String name,
BeanDefinitionRegistry registry) {
registerChannelBeanDefinition(Input.class, qualifierValue, name, registry);
}
public static void registerOutputChannelBeanDefinition(String name,
BeanDefinitionRegistry registry) {
registerOutputChannelBeanDefinition(DEFAULT_OUTPUT_QUALIFIER_VALUE, name, registry);
}
public static void registerOutputChannelBeanDefinition(String qualifierValue, String name,BeanDefinitionRegistry registry) {
registerChannelBeanDefinition(Output.class, qualifierValue, name, registry);
}
private static void registerChannelBeanDefinition(
Class<? extends Annotation> qualifier, String qualifierValue, String name,
BeanDefinitionRegistry registry) {
RootBeanDefinition rootBeanDefinition = new RootBeanDefinition(
DirectChannelFactoryBean.class);
rootBeanDefinition.addQualifier(new AutowireCandidateQualifier(qualifier, qualifierValue));
registry.registerBeanDefinition(name, rootBeanDefinition);
}
public static List<String> registerChannelBeanDefinitions(Class<?> type,
final BeanDefinitionRegistry registry) {
final List<String> channelNames = new ArrayList<>();
ReflectionUtils.doWithMethods(type, new MethodCallback() {
@Override
public void doWith(Method method) throws IllegalArgumentException,
IllegalAccessException {
Input input = AnnotationUtils.findAnnotation(method, Input.class);
if (input != null) {
String name = getName(input, method);
registerInputChannelBeanDefinition(input.value(), name, registry);
channelNames.add(name);
}
Output output = AnnotationUtils.findAnnotation(method, Output.class);
if (output != null) {
String name = getName(output, method);
registerOutputChannelBeanDefinition(output.value(), name, registry);
channelNames.add(name);
}
}
});
return channelNames;
}
public static void registerChannelsQualifiedBeanDefinitions(Class<?> parent, Class<?> type,
final BeanDefinitionRegistry registry) {
if (type.isInterface()) {
RootBeanDefinition rootBeanDefinition = new RootBeanDefinition(
ChannelProxyFactory.class);
rootBeanDefinition.addQualifier(new AutowireCandidateQualifier(ModuleChannels.class, parent));
rootBeanDefinition.getConstructorArgumentValues().addGenericArgumentValue(
type);
registry.registerBeanDefinition(type.getName(), rootBeanDefinition);
}
else {
RootBeanDefinition rootBeanDefinition = new RootBeanDefinition(type);
rootBeanDefinition.addQualifier(new AutowireCandidateQualifier(ModuleChannels.class, parent));
registry.registerBeanDefinition(type.getName(), rootBeanDefinition);
}
}
private static String getName(Annotation annotation, Method method) {
Map<String, Object> attrs = AnnotationUtils.getAnnotationAttributes(annotation,
false);
if (attrs.containsKey("value") && StringUtils.hasText((CharSequence) attrs.get("value"))) {
return (String) attrs.get("value");
}
return method.getName();
}
static class ChannelProxyFactory implements MethodInterceptor,
FactoryBean<Object>, BeanFactoryAware {
private Class<?> type;
private Object value = null;
private BeanFactory beanFactory;
public ChannelProxyFactory(Class<?> type) {
this.type = type;
}
@Override
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
this.beanFactory = beanFactory;
}
@Override
public Object invoke(MethodInvocation invocation) throws Throwable {
Method method = invocation.getMethod();
if (MessageChannel.class.isAssignableFrom(method.getReturnType())) {
Input input = AnnotationUtils.findAnnotation(method, Input.class);
if (input != null) {
String name = getName(input, method);
return this.beanFactory.getBean(name);
}
Output output = AnnotationUtils.findAnnotation(method, Output.class);
if (output != null) {
String name = getName(output, method);
return this.beanFactory.getBean(name);
}
}
return null;
}
@Override
public Object getObject() throws Exception {
if (this.value == null) {
this.value = create();
}
return this.value;
}
private Object create() {
ProxyFactory factory = new ProxyFactory(this.type, this);
return factory.getProxy();
}
@Override
public Class<?> getObjectType() {
return this.type;
}
@Override
public boolean isSingleton() {
return true;
}
}
}

61
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/adapter/DefaultChannelLocatorTests.java
View File

@@ -1,61 +0,0 @@
/*
* Copyright 2015 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.adapter;
import static org.junit.Assert.assertEquals;
import java.util.Collections;
import org.junit.Test;
import org.springframework.cloud.stream.adapter.DefaultChannelLocator;
import org.springframework.cloud.stream.config.ChannelBindingProperties;
/**
* @author Dave Syer
*
*/
public class DefaultChannelLocatorTests {
private ChannelBindingProperties module = new ChannelBindingProperties();
private DefaultChannelLocator locator = new DefaultChannelLocator(this.module);
@Test
public void oneOutput() throws Exception {
assertEquals("output", this.locator.locate("output"));
}
@Test
public void oneOutputWithShortcutPath() throws Exception {
module.getBindings().put("outputWithShortcutPath","group.0.shortcut");
assertEquals("group.0.shortcut", this.locator.locate("outputWithShortcutPath"));
}
@Test
public void oneOutputWithFullPath() throws Exception {
module.getBindings().put("outputWithFullPath", Collections.singletonMap(ChannelBindingProperties.PATH,"group.0.full"));
assertEquals("group.0.full", this.locator.locate("outputWithFullPath"));
}
@Test
public void oneOutputTopic() throws Exception {
module.getBindings().put("outputWithTopic", Collections.singletonMap(ChannelBindingProperties.PATH,"topic:group.0"));
assertEquals("topic:group.0", this.locator.locate("outputWithTopic"));
}
}

58
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/aggregation/ModuleAggregationTest.java Normal file
View File

@@ -0,0 +1,58 @@
/*
* Copyright 2015 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.aggregation;
import static org.hamcrest.collection.IsCollectionWithSize.hasSize;
import static org.junit.Assert.assertThat;
import org.junit.Test;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.cloud.stream.aggregate.AggregateApplication;
import org.springframework.cloud.stream.aggregate.SharedChannelRegistry;
import org.springframework.cloud.stream.annotation.EnableModule;
import org.springframework.cloud.stream.annotation.Processor;
import org.springframework.cloud.stream.annotation.Source;
import org.springframework.context.ConfigurableApplicationContext;
/**
* @author Marius Bogoevici
*/
public class ModuleAggregationTest {
@Test
public void testModuleAggregation() {
ConfigurableApplicationContext aggregatedApplicationContext = AggregateApplication.run(TestSource.class,
TestProcessor.class);
SharedChannelRegistry sharedChannelRegistry = aggregatedApplicationContext.getBean(SharedChannelRegistry.class);
assertThat(sharedChannelRegistry.getAll().keySet(), hasSize(2));
}
@EnableModule(Source.class)
@EnableAutoConfiguration
public static class TestSource {
}
@EnableModule(Processor.class)
@EnableAutoConfiguration
public static class TestProcessor {
}
}

4
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ArbitraryInterfaceBindingTestsWithBindingTargets.java
View File

@@ -5,7 +5,7 @@
* 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
* 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,
@@ -47,9 +47,11 @@ public class ArbitraryInterfaceBindingTestsWithBindingTargets {
@ModuleChannels(ArbitraryInterfaceBindingTestsWithBindingTargets.TestFooChannels.class)
public FooChannels fooChannels;
@SuppressWarnings("rawtypes")
@Autowired
private Binder binder;
@SuppressWarnings("unchecked")
@Test
public void testArbitraryInterfaceChannelsBound() {
verify(binder).bindConsumer(eq("someQueue.0"), eq(fooChannels.foo()), Mockito.<Properties>any());

4
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ArbitraryInterfaceBindingTestsWithDefaults.java
View File

@@ -5,7 +5,7 @@
* 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
* 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,
@@ -46,9 +46,11 @@ public class ArbitraryInterfaceBindingTestsWithDefaults {
@ModuleChannels(ArbitraryInterfaceBindingTestsWithDefaults.TestFooChannels.class)
public FooChannels fooChannels;
@SuppressWarnings("rawtypes")
@Autowired
private Binder binder;
@SuppressWarnings("unchecked")
@Test
public void testArbitraryInterfaceChannelsBound() {
verify(binder).bindConsumer(eq("foo"), eq(fooChannels.foo()), Mockito.<Properties>any());

3
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderAwareChannelResolverTests.java
View File

@@ -36,6 +36,7 @@ import org.junit.Test;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.cloud.stream.binder.local.LocalMessageChannelBinder;
import org.springframework.cloud.stream.binding.BinderAwareChannelResolver;
import org.springframework.context.support.StaticApplicationContext;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.PublishSubscribeChannel;
@@ -155,9 +156,11 @@ public class BinderAwareChannelResolverTests {
@Test
public void propertyPassthrough() {
Properties properties = new Properties();
@SuppressWarnings("rawtypes")
Binder binder = mock(Binder.class);
doReturn(new DirectChannel()).when(binder).bindDynamicProducer("queue:foo", properties);
doReturn(new DirectChannel()).when(binder).bindDynamicPubSubProducer("topic:bar", properties);
@SuppressWarnings("unchecked")
BinderAwareChannelResolver resolver = new BinderAwareChannelResolver(binder, properties);
BeanFactory beanFactory = new DefaultListableBeanFactory();
resolver.setBeanFactory(beanFactory);

2
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/FooChannels.java
View File

@@ -5,7 +5,7 @@
* 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
* 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,

4
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ProcessorBindingTestsWithBindingTargets.java
View File

@@ -5,7 +5,7 @@
* 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
* 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,
@@ -43,12 +43,14 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@SpringApplicationConfiguration(ProcessorBindingTestsWithBindingTargets.TestProcessor.class)
public class ProcessorBindingTestsWithBindingTargets {
@SuppressWarnings("rawtypes")
@Autowired
private Binder binder;
@Autowired @ModuleChannels(TestProcessor.class)
private Processor testProcessor;
@SuppressWarnings("unchecked")
@Test
public void testSourceOutputChannelBound() {
verify(binder).bindConsumer(eq("testtock.0"), eq(testProcessor.input()), Mockito.<Properties>any());

5
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ProcessorBindingTestsWithDefaults.java
View File

@@ -5,7 +5,7 @@
* 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
* 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,
@@ -21,7 +21,6 @@ import static org.mockito.Mockito.verifyNoMoreInteractions;
import java.util.Properties;
import org.apache.catalina.core.ApplicationContext;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mockito;
@@ -43,12 +42,14 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@SpringApplicationConfiguration(ProcessorBindingTestsWithDefaults.TestProcessor.class)
public class ProcessorBindingTestsWithDefaults {
@SuppressWarnings("rawtypes")
@Autowired
private Binder binder;
@Autowired @ModuleChannels(TestProcessor.class)
private Processor processor;
@SuppressWarnings("unchecked")
@Test
public void testSourceOutputChannelBound() {
Mockito.verify(binder).bindConsumer(eq("input"), eq(processor.input()), Mockito.<Properties>any());

4
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SinkBindingTestsWithBindingTargets.java
View File

@@ -5,7 +5,7 @@
* 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
* 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,
@@ -44,12 +44,14 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@SpringApplicationConfiguration(SinkBindingTestsWithBindingTargets.TestSink.class)
public class SinkBindingTestsWithBindingTargets {
@SuppressWarnings("rawtypes")
@Autowired
private Binder binder;
@Autowired @ModuleChannels(TestSink.class)
private Sink testSink;
@SuppressWarnings("unchecked")
@Test
public void testSourceOutputChannelBound() {
verify(binder).bindConsumer(eq("testtock"), eq(testSink.input()), Mockito.<Properties>any());

5
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SinkBindingTestsWithDefaults.java
View File

@@ -5,7 +5,7 @@
* 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
* 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,
@@ -31,7 +31,6 @@ import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.test.SpringApplicationConfiguration;
import org.springframework.cloud.stream.annotation.EnableModule;
import org.springframework.cloud.stream.annotation.ModuleChannels;
import org.springframework.cloud.stream.annotation.Processor;
import org.springframework.cloud.stream.annotation.Sink;
import org.springframework.cloud.stream.utils.MockBinderConfiguration;
import org.springframework.context.annotation.Import;
@@ -44,12 +43,14 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@SpringApplicationConfiguration(SinkBindingTestsWithDefaults.TestSink.class)
public class SinkBindingTestsWithDefaults {
@SuppressWarnings("rawtypes")
@Autowired
private Binder binder;
@Autowired @ModuleChannels(TestSink.class)
private Sink testSink;
@SuppressWarnings("unchecked")
@Test
public void testSourceOutputChannelBound() {
verify(binder).bindConsumer(eq("input"), eq(testSink.input()), Mockito.<Properties>any());

4
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SourceBindingTestsWithBindingTargets.java
View File

@@ -5,7 +5,7 @@
* 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
* 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,
@@ -44,12 +44,14 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@SpringApplicationConfiguration(SourceBindingTestsWithBindingTargets.TestSource.class)
public class SourceBindingTestsWithBindingTargets {
@SuppressWarnings("rawtypes")
@Autowired
private Binder binder;
@Autowired @ModuleChannels(TestSource.class)
private Source testSource;
@SuppressWarnings("unchecked")
@Test
public void testSourceOutputChannelBound() {
verify(binder).bindProducer(eq("testtock"), eq(testSource.output()), Mockito.<Properties>any());

4
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SourceBindingTestsWithDefaults.java
View File

@@ -5,7 +5,7 @@
* 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
* 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,
@@ -43,12 +43,14 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@SpringApplicationConfiguration(SourceBindingTestsWithDefaults.TestSource.class)
public class SourceBindingTestsWithDefaults {
@SuppressWarnings("rawtypes")
@Autowired
private Binder binder;
@Autowired @ModuleChannels(TestSource.class)
private Source testSource;
@SuppressWarnings("unchecked")
@Test
public void testSourceOutputChannelBound() {
verify(binder).bindProducer(eq("output"), eq(testSource.output()), Mockito.<Properties>any());

145
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/ChannelBindingAdapterConfigurationTests.java
View File

@@ -1,145 +0,0 @@
/*
* Copyright 2015 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.config;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.boot.autoconfigure.PropertyPlaceholderAutoConfiguration;
import org.springframework.boot.test.SpringApplicationConfiguration;
import org.springframework.cloud.stream.adapter.ChannelBinding;
import org.springframework.cloud.stream.adapter.ChannelBindingAdapter;
import org.springframework.cloud.stream.adapter.OutputChannelBinding;
import org.springframework.cloud.stream.binder.local.LocalMessageChannelBinder;
import org.springframework.cloud.stream.config.ChannelBindingAdapterConfigurationTests.Empty;
import org.springframework.cloud.stream.utils.MessageChannelBeanDefinitionRegistryUtils;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.annotation.DirtiesContext.ClassMode;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Dave Syer
*/
@RunWith(SpringJUnit4ClassRunner.class)
@SpringApplicationConfiguration(classes = Empty.class)
@DirtiesContext(classMode = ClassMode.AFTER_EACH_TEST_METHOD)
public class ChannelBindingAdapterConfigurationTests {
@Autowired
private DefaultListableBeanFactory context;
@Autowired
private ChannelBindingAdapter adapter;
@Autowired
private ChannelBindingAdapterConfiguration configuration;
@Autowired
private ChannelBindingProperties module;
@Before
public void init() {
}
@Test
public void oneOutput() throws Exception {
MessageChannelBeanDefinitionRegistryUtils.registerOutputChannelBeanDefinition("output", context);
refresh();
Collection<OutputChannelBinding> channels = this.adapter.getChannelsMetadata().getOutputChannels();
assertEquals(1, channels.size());
assertEquals("output", channels.iterator().next().getRemoteName());
assertEquals("tap:output", channels.iterator().next().getTapChannelName());
}
private void refresh() {
Collection<OutputChannelBinding> channels = this.configuration.getOutputChannels();
for (OutputChannelBinding channel : channels) {
channel.setTapped(true);
}
this.adapter.stop();
this.adapter.setOutputChannels(channels);
this.adapter.start();
}
@Test
public void twoOutputs() throws Exception {
MessageChannelBeanDefinitionRegistryUtils.registerOutputChannelBeanDefinition("output", context);
MessageChannelBeanDefinitionRegistryUtils.registerOutputChannelBeanDefinition("foo", context);
module.getBindings().put("output", "group.0");
module.getBindings().put("foo", "topic:group.0");
refresh();
Collection<OutputChannelBinding> channels = this.adapter.getChannelsMetadata().getOutputChannels();
List<String> remoteBindingNames = getRemoteBindingNames(channels);
assertEquals(2, channels.size());
assertTrue(remoteBindingNames.contains("group.0"));
assertTrue(remoteBindingNames.contains("topic:group.0"));
}
private List<String> getRemoteBindingNames(Collection<? extends ChannelBinding> channels) {
List<String> list = new ArrayList<String>();
for (ChannelBinding binding : channels) {
list.add(binding.getRemoteName());
}
return list;
}
@Test
public void overrideNaturalOutputChannelName() throws Exception {
MessageChannelBeanDefinitionRegistryUtils.registerOutputChannelBeanDefinition("output", context);
module.getBindings().put("output", "group.0");
refresh();
Collection<OutputChannelBinding> channels = this.adapter.getChannelsMetadata().getOutputChannels();
assertEquals(1, channels.size());
assertEquals("group.0", channels.iterator().next().getRemoteName());
assertEquals("tap:group.0", channels.iterator().next().getTapChannelName());
}
@Test
public void overrideNaturalOutputChannelNamedQueueWithTopic() throws Exception {
MessageChannelBeanDefinitionRegistryUtils.registerOutputChannelBeanDefinition("output", context);
module.getBindings().put("output", "topic:group.0");
refresh();
Collection<OutputChannelBinding> channels = this.adapter.getChannelsMetadata().getOutputChannels();
assertEquals(1, channels.size());
assertEquals("topic:group.0", channels.iterator().next().getRemoteName());
// is this correct?
assertEquals("tap:group.0", channels.iterator().next().getTapChannelName());
}
@Configuration
@Import({ChannelBindingAdapterConfiguration.class, PropertyPlaceholderAutoConfiguration.class})
protected static class Empty {
@Bean
public LocalMessageChannelBinder binder() {
return new LocalMessageChannelBinder();
}
}
}

4
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/interceptor/BoundChannelsInterceptedTest.java
View File

@@ -5,7 +5,7 @@
* 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
* 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,
@@ -22,8 +22,6 @@ import static org.mockito.Mockito.verifyNoMoreInteractions;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mockito;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.test.SpringApplicationConfiguration;

2
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/utils/MockBinderConfiguration.java
View File

@@ -5,7 +5,7 @@
* 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
* 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,