diff --git a/.gitignore b/.gitignore index 7087c8692..eade3d4c5 100644 --- a/.gitignore +++ b/.gitignore @@ -21,3 +21,5 @@ _site/ .factorypath spring-xd-samples/*/xd dump.rdb +.apt_generated +artifacts diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-local/src/main/java/org/springframework/cloud/stream/binder/local/config/LocalExecutorConfigurationProperties.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-local/src/main/java/org/springframework/cloud/stream/binder/local/config/LocalExecutorConfigurationProperties.java index 79e624e09..95ed90f1c 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-local/src/main/java/org/springframework/cloud/stream/binder/local/config/LocalExecutorConfigurationProperties.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-local/src/main/java/org/springframework/cloud/stream/binder/local/config/LocalExecutorConfigurationProperties.java @@ -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; } } diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-local/src/main/java/org/springframework/cloud/stream/binder/local/config/LocalMessageChannelBinderConfiguration.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-local/src/main/java/org/springframework/cloud/stream/binder/local/config/LocalMessageChannelBinderConfiguration.java index c4e043e75..1108a8a91 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-local/src/main/java/org/springframework/cloud/stream/binder/local/config/LocalMessageChannelBinderConfiguration.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-local/src/main/java/org/springframework/cloud/stream/binder/local/config/LocalMessageChannelBinderConfiguration.java @@ -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(); diff --git a/spring-cloud-stream-module-launcher/.gitignore b/spring-cloud-stream-module-launcher/.gitignore deleted file mode 100644 index 44898dfcf..000000000 --- a/spring-cloud-stream-module-launcher/.gitignore +++ /dev/null @@ -1 +0,0 @@ -artifacts \ No newline at end of file diff --git a/spring-cloud-stream-module-launcher/src/main/java/org/springframework/boot/loader/ModuleJarLauncher.java b/spring-cloud-stream-module-launcher/src/main/java/org/springframework/boot/loader/ModuleJarLauncher.java index c580d750e..5bb462a38 100644 --- a/spring-cloud-stream-module-launcher/src/main/java/org/springframework/boot/loader/ModuleJarLauncher.java +++ b/spring-cloud-stream-module-launcher/src/main/java/org/springframework/boot/loader/ModuleJarLauncher.java @@ -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); } } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/adapter/ChannelBinding.java b/spring-cloud-stream-module-launcher/src/main/java/org/springframework/cloud/stream/module/launcher/ArchiveMatchingEntryFilter.java similarity index 50% rename from spring-cloud-stream/src/main/java/org/springframework/cloud/stream/adapter/ChannelBinding.java rename to spring-cloud-stream-module-launcher/src/main/java/org/springframework/cloud/stream/module/launcher/ArchiveMatchingEntryFilter.java index 3a59fb4a9..419c8597a 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/adapter/ChannelBinding.java +++ b/spring-cloud-stream-module-launcher/src/main/java/org/springframework/cloud/stream/module/launcher/ArchiveMatchingEntryFilter.java @@ -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; - } - } diff --git a/spring-cloud-stream-module-launcher/src/main/java/org/springframework/cloud/stream/module/launcher/ModuleLauncher.java b/spring-cloud-stream-module-launcher/src/main/java/org/springframework/cloud/stream/module/launcher/ModuleLauncher.java index 31999202a..0c38f36b5 100644 --- a/spring-cloud-stream-module-launcher/src/main/java/org/springframework/cloud/stream/module/launcher/ModuleLauncher.java +++ b/spring-cloud-stream-module-launcher/src/main/java/org/springframework/cloud/stream/module/launcher/ModuleLauncher.java @@ -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 { * <groupId>:<artifactId>[:<extension>[:<classifier>]]:<version> * * @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 moduleLaunchRequests) { + public void launch(List moduleLaunchRequests, boolean aggregate, String parentArgs[]) { List 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 moduleLaunchRequests, boolean aggregate) { + this.launch(moduleLaunchRequests, aggregate, new String[0]); + } + + public void launch(List moduleLaunchRequests) { + this.launch(moduleLaunchRequests, false); } /** @@ -92,6 +121,56 @@ public class ModuleLauncher { return result; } + public void launchAggregatedModules(List moduleLaunchRequests, final String[] parentArgs) { + try { + List mainClassNames = new ArrayList<>(); + List jarURLs = new ArrayList<>(); + List seenArchives = new ArrayList<>(); + final List 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> 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 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> mainClasses; + + private final List arguments; + + public ModuleAggregatorRunner(ClassLoader classLoader, List> mainClasses, String[] parentArgs, List 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); + } + } + } } diff --git a/spring-cloud-stream-module-launcher/src/main/java/org/springframework/cloud/stream/module/launcher/ModuleLauncherProperties.java b/spring-cloud-stream-module-launcher/src/main/java/org/springframework/cloud/stream/module/launcher/ModuleLauncherProperties.java index 21100e1ea..3c7b8a237 100644 --- a/spring-cloud-stream-module-launcher/src/main/java/org/springframework/cloud/stream/module/launcher/ModuleLauncherProperties.java +++ b/spring-cloud-stream-module-launcher/src/main/java/org/springframework/cloud/stream/module/launcher/ModuleLauncherProperties.java @@ -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> 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; diff --git a/spring-cloud-stream-module-launcher/src/main/java/org/springframework/cloud/stream/module/launcher/ModuleLauncherRunner.java b/spring-cloud-stream-module-launcher/src/main/java/org/springframework/cloud/stream/module/launcher/ModuleLauncherRunner.java index 1b5a2850a..09790202b 100644 --- a/spring-cloud-stream-module-launcher/src/main/java/org/springframework/cloud/stream/module/launcher/ModuleLauncherRunner.java +++ b/spring-cloud-stream-module-launcher/src/main/java/org/springframework/cloud/stream/module/launcher/ModuleLauncherRunner.java @@ -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 generateModuleLaunchRequests() { @@ -71,4 +73,15 @@ public class ModuleLauncherRunner implements CommandLineRunner { } return requests; } + + private String[] extractAggregateProperties(String[] args) { + List filteredProperties = new ArrayList<>(); + for (String arg : args) { + if (arg.startsWith("--spring.")) { + filteredProperties.add(arg); + } + } + return filteredProperties.toArray(new String[filteredProperties.size()]); + } + } diff --git a/spring-cloud-stream-module-launcher/src/main/java/org/springframework/cloud/stream/module/resolver/ModuleResolver.java b/spring-cloud-stream-module-launcher/src/main/java/org/springframework/cloud/stream/module/resolver/ModuleResolver.java index 827080173..b9bc712b8 100644 --- a/spring-cloud-stream-module-launcher/src/main/java/org/springframework/cloud/stream/module/resolver/ModuleResolver.java +++ b/spring-cloud-stream-module-launcher/src/main/java/org/springframework/cloud/stream/module/resolver/ModuleResolver.java @@ -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); } diff --git a/spring-cloud-stream-module-launcher/src/main/java/org/springframework/cloud/stream/module/resolver/ModuleResolverProperties.java b/spring-cloud-stream-module-launcher/src/main/java/org/springframework/cloud/stream/module/resolver/ModuleResolverProperties.java index 434c99516..0036a0412 100644 --- a/spring-cloud-stream-module-launcher/src/main/java/org/springframework/cloud/stream/module/resolver/ModuleResolverProperties.java +++ b/spring-cloud-stream-module-launcher/src/main/java/org/springframework/cloud/stream/module/resolver/ModuleResolverProperties.java @@ -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, diff --git a/spring-cloud-stream-module-launcher/src/main/java/org/springframework/cloud/stream/module/utils/ClassloaderUtils.java b/spring-cloud-stream-module-launcher/src/main/java/org/springframework/cloud/stream/module/utils/ClassloaderUtils.java new file mode 100644 index 000000000..c45098d02 --- /dev/null +++ b/spring-cloud-stream-module-launcher/src/main/java/org/springframework/cloud/stream/module/utils/ClassloaderUtils.java @@ -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; + } +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/adapter/ChannelLocator.java b/spring-cloud-stream-samples/double/src/main/java/config/sink/SinkApplication.java similarity index 78% rename from spring-cloud-stream/src/main/java/org/springframework/cloud/stream/adapter/ChannelLocator.java rename to spring-cloud-stream-samples/double/src/main/java/config/sink/SinkApplication.java index 88e53fe29..e093d1ff9 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/adapter/ChannelLocator.java +++ b/spring-cloud-stream-samples/double/src/main/java/config/sink/SinkApplication.java @@ -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 { } diff --git a/spring-cloud-stream-samples/double/src/main/java/config/SinkModuleDefinition.java b/spring-cloud-stream-samples/double/src/main/java/config/sink/SinkModuleDefinition.java similarity index 98% rename from spring-cloud-stream-samples/double/src/main/java/config/SinkModuleDefinition.java rename to spring-cloud-stream-samples/double/src/main/java/config/sink/SinkModuleDefinition.java index 5276a61bf..6f6d4002a 100644 --- a/spring-cloud-stream-samples/double/src/main/java/config/SinkModuleDefinition.java +++ b/spring-cloud-stream-samples/double/src/main/java/config/sink/SinkModuleDefinition.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package config; +package config.sink; import org.slf4j.Logger; import org.slf4j.LoggerFactory; diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/adapter/InputChannelBinding.java b/spring-cloud-stream-samples/double/src/main/java/config/source/SourceApplication.java similarity index 71% rename from spring-cloud-stream/src/main/java/org/springframework/cloud/stream/adapter/InputChannelBinding.java rename to spring-cloud-stream-samples/double/src/main/java/config/source/SourceApplication.java index 74a386622..f1d907786 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/adapter/InputChannelBinding.java +++ b/spring-cloud-stream-samples/double/src/main/java/config/source/SourceApplication.java @@ -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 { } diff --git a/spring-cloud-stream-samples/double/src/main/java/config/SourceModuleDefinition.java b/spring-cloud-stream-samples/double/src/main/java/config/source/SourceModuleDefinition.java similarity index 86% rename from spring-cloud-stream-samples/double/src/main/java/config/SourceModuleDefinition.java rename to spring-cloud-stream-samples/double/src/main/java/config/source/SourceModuleDefinition.java index a3c6c6d6b..8d04e6a4f 100644 --- a/spring-cloud-stream-samples/double/src/main/java/config/SourceModuleDefinition.java +++ b/spring-cloud-stream-samples/double/src/main/java/config/source/SourceModuleDefinition.java @@ -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 timerMessageSource() { return () -> new GenericMessage<>(new SimpleDateFormat(this.format).format(new Date())); } diff --git a/spring-cloud-stream-samples/double/src/main/java/demo/DoubleApplication.java b/spring-cloud-stream-samples/double/src/main/java/demo/DoubleApplication.java index e66b398f7..3ebe00bd1 100644 --- a/spring-cloud-stream-samples/double/src/main/java/demo/DoubleApplication.java +++ b/spring-cloud-stream-samples/double/src/main/java/demo/DoubleApplication.java @@ -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); } } diff --git a/spring-cloud-stream-samples/double/src/main/resources/application.yml b/spring-cloud-stream-samples/double/src/main/resources/application.yml index 136d06384..d84e8d13d 100644 --- a/spring-cloud-stream-samples/double/src/main/resources/application.yml +++ b/spring-cloud-stream-samples/double/src/main/resources/application.yml @@ -1 +1 @@ - \ No newline at end of file +fixedDelay: 1000 diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/adapter/ChannelBindingAdapter.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/adapter/ChannelBindingAdapter.java deleted file mode 100644 index 9f6aa939b..000000000 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/adapter/ChannelBindingAdapter.java +++ /dev/null @@ -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 binder; - - private MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory(); - - private Collection outputChannels = Collections.emptySet(); - - private Collection 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 channelResolver; - - private Map bindings = new HashMap(); - - public ChannelBindingAdapter(ChannelBindingProperties module, Binder 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 channelResolver) { - this.channelResolver = channelResolver; - } - - public void setMessageBuilderFactory(MessageBuilderFactory messageBuilderFactory) { - this.messageBuilderFactory = messageBuilderFactory; - } - - public void setTrackHistory(boolean trackHistory) { - this.trackHistory = trackHistory; - } - - public void setOutputChannels(Collection outputChannels) { - this.outputChannels = new LinkedHashSet(outputChannels); - } - - public void setInputChannels(Collection inputChannels) { - this.inputChannels = new LinkedHashSet(inputChannels); - } - - public ChannelsMetadata getChannelsMetadata() { - ChannelsMetadata channels = new ChannelsMetadata(); - channels.setModule(this.module); - channels.setInputChannels(new LinkedHashSet(this.inputChannels)); - channels.setOutputChannels(new LinkedHashSet(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 historyProperties = new LinkedHashMap(); - 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 historyProps) { - if (channel instanceof ChannelInterceptorAware) { - ((ChannelInterceptorAware) channel) - .addInterceptor(new ChannelInterceptorAdapter() { - - @Override - public Message preSend(Message message, MessageChannel channel) { - @SuppressWarnings("unchecked") - Collection> history = (Collection>) message - .getHeaders().get(BinderHeaders.BINDER_HISTORY); - if (history == null) { - history = new ArrayList>(1); - } - else { - history = new ArrayList>(history); - } - Map map = new LinkedHashMap(); - 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; - } -} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/adapter/ChannelsMetadata.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/adapter/ChannelsMetadata.java deleted file mode 100644 index bb1d4db77..000000000 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/adapter/ChannelsMetadata.java +++ /dev/null @@ -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 outputChannels = Collections.emptySet(); - private Collection inputChannels = Collections.emptySet(); - private ChannelBindingProperties module; - - public ChannelBindingProperties getModule() { - return this.module; - } - - public void setModule(ChannelBindingProperties module) { - this.module = module; - } - - public Collection getOutputChannels() { - return this.outputChannels; - } - - public void setOutputChannels(Collection outputChannels) { - this.outputChannels = outputChannels; - } - - public Collection getInputChannels() { - return this.inputChannels; - } - - public void setInputChannels(Collection inputChannels) { - this.inputChannels = inputChannels; - } - -} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/adapter/DefaultChannelLocator.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/adapter/DefaultChannelLocator.java deleted file mode 100644 index 6df6f2a29..000000000 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/adapter/DefaultChannelLocator.java +++ /dev/null @@ -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; - } - -} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/adapter/OutputChannelBinding.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/adapter/OutputChannelBinding.java deleted file mode 100644 index 99f5b7a5b..000000000 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/adapter/OutputChannelBinding.java +++ /dev/null @@ -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; - } - -} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/AggregateApplication.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/AggregateApplication.java new file mode 100644 index 000000000..3c6505ac0 --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/AggregateApplication.java @@ -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(); + } + } + +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/AggregateBuilder.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/AggregateBuilder.java index 663374714..2091d2386 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/AggregateBuilder.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/AggregateBuilder.java @@ -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; diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/SharedChannelRegistry.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/SharedChannelRegistry.java new file mode 100644 index 000000000..e00361aad --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/SharedChannelRegistry.java @@ -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 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 getAll() { + return Collections.unmodifiableMap(sharedChannels); + } +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/EnableModule.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/EnableModule.java index 5830dad29..22cf438e6 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/EnableModule.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/EnableModule.java @@ -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 {}; diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/Bindable.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/Bindable.java new file mode 100644 index 000000000..5b62928ef --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/Bindable.java @@ -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); + +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindableProxyFactory.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindableProxyFactory.java new file mode 100644 index 000000000..05f09c313 --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindableProxyFactory.java @@ -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, + 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 inputs = new HashMap<>(); + + private Map 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 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 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 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 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; + } + } + +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderAwareChannelResolver.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BinderAwareChannelResolver.java similarity index 95% rename from spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderAwareChannelResolver.java rename to spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BinderAwareChannelResolver.java index d433f09ab..136053612 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderAwareChannelResolver.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BinderAwareChannelResolver.java @@ -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; diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderAwareRouterBeanPostProcessor.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BinderAwareRouterBeanPostProcessor.java similarity index 97% rename from spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderAwareRouterBeanPostProcessor.java rename to spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BinderAwareRouterBeanPostProcessor.java index 511ba4c87..b01a70bd9 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderAwareRouterBeanPostProcessor.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BinderAwareRouterBeanPostProcessor.java @@ -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; diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingBeanDefinitionRegistryUtils.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingBeanDefinitionRegistryUtils.java new file mode 100644 index 000000000..615cc5fbb --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingBeanDefinitionRegistryUtils.java @@ -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 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 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 attrs = AnnotationUtils.getAnnotationAttributes(annotation, + false); + if (attrs.containsKey("value") + && StringUtils.hasText((CharSequence) attrs.get("value"))) { + return (String) attrs.get("value"); + } + return method.getName(); + } + +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ChannelBindingLifecycle.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ChannelBindingLifecycle.java new file mode 100644 index 000000000..c9b38ffb3 --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ChannelBindingLifecycle.java @@ -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 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 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; + } +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ChannelBindingService.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ChannelBindingService.java new file mode 100644 index 000000000..8c1dced5a --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ChannelBindingService.java @@ -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 binder; + + private ChannelBindingProperties channelBindingProperties; + + public ChannelBindingService(ChannelBindingProperties channelBindingProperties, + Binder 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); + } + + +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ModuleRegistrar.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingBeansRegistrar.java similarity index 57% rename from spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ModuleRegistrar.java rename to spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingBeansRegistrar.java index 1077df811..5eb619a47 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ModuleRegistrar.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingBeansRegistrar.java @@ -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 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> collectClasses(List list) { diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingAdapterConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingAdapterConfiguration.java deleted file mode 100644 index fa4c8db7e..000000000 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingAdapterConfiguration.java +++ /dev/null @@ -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 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 getOutputChannels() { - Set 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 getInputChannels() { - Set 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() { - - @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); - } - - }); - } - - } - -} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingProperties.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingProperties.java index 273792b6d..82668942b 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingProperties.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingProperties.java @@ -87,5 +87,5 @@ public class ChannelBindingProperties { public String getTapChannelName(String channelName) { return "tap:" + getBindingPath(channelName); } - + } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceConfiguration.java new file mode 100644 index 000000000..ab0a0e176 --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceConfiguration.java @@ -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 binder) { + return new ChannelBindingService(channelBindingProperties, binder); + } + + @Bean + @DependsOn("bindingService") + public ChannelBindingLifecycle channelBindingLifecycle() { + return new ChannelBindingLifecycle(); + } + + @Bean + public BinderAwareChannelResolver binderAwareChannelResolver( + Binder 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() { + + @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); + } + + }); + } + + } + +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/DirectChannelFactoryBean.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/DirectChannelFactoryBean.java deleted file mode 100644 index 55224231f..000000000 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/DirectChannelFactoryBean.java +++ /dev/null @@ -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, 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; - } -} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/endpoint/ChannelsEndpoint.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/endpoint/ChannelsEndpoint.java deleted file mode 100644 index 8d1c98a46..000000000 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/endpoint/ChannelsEndpoint.java +++ /dev/null @@ -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> { - - private ChannelBindingAdapter adapter; - - public ChannelsEndpoint(ChannelBindingAdapter adapter) { - super("channels"); - this.adapter = adapter; - } - - @RequestMapping(value="/channels/taps") - public List taps() { - List list = new ArrayList(); - 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 invoke() { - LinkedHashMap map = new LinkedHashMap(); - ChannelsMetadata channels = adapter.getChannelsMetadata(); - map.put("inputChannels", channels.getInputChannels()); - map.put("outputChannels", channels.getOutputChannels()); - map.put("module", channels.getModule()); - return map; - } - -} \ No newline at end of file diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/utils/MessageChannelBeanDefinitionRegistryUtils.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/utils/MessageChannelBeanDefinitionRegistryUtils.java deleted file mode 100644 index 2ec6bc8d0..000000000 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/utils/MessageChannelBeanDefinitionRegistryUtils.java +++ /dev/null @@ -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 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 registerChannelBeanDefinitions(Class type, - final BeanDefinitionRegistry registry) { - final List 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 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, 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; - } - - } - -} diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/adapter/DefaultChannelLocatorTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/adapter/DefaultChannelLocatorTests.java deleted file mode 100644 index ad2ed0526..000000000 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/adapter/DefaultChannelLocatorTests.java +++ /dev/null @@ -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")); - } - -} diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/aggregation/ModuleAggregationTest.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/aggregation/ModuleAggregationTest.java new file mode 100644 index 000000000..f7af9b0bf --- /dev/null +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/aggregation/ModuleAggregationTest.java @@ -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 { + + } + +} diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ArbitraryInterfaceBindingTestsWithBindingTargets.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ArbitraryInterfaceBindingTestsWithBindingTargets.java index 2089d5a0b..158e4a28a 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ArbitraryInterfaceBindingTestsWithBindingTargets.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ArbitraryInterfaceBindingTestsWithBindingTargets.java @@ -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.any()); diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ArbitraryInterfaceBindingTestsWithDefaults.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ArbitraryInterfaceBindingTestsWithDefaults.java index 280150980..b02b97950 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ArbitraryInterfaceBindingTestsWithDefaults.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ArbitraryInterfaceBindingTestsWithDefaults.java @@ -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.any()); diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderAwareChannelResolverTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderAwareChannelResolverTests.java index 2a99fdbaf..56b9e2b2f 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderAwareChannelResolverTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderAwareChannelResolverTests.java @@ -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); diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/FooChannels.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/FooChannels.java index 341307961..a96f812a9 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/FooChannels.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/FooChannels.java @@ -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, diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ProcessorBindingTestsWithBindingTargets.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ProcessorBindingTestsWithBindingTargets.java index cf39fb729..b4dd811f5 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ProcessorBindingTestsWithBindingTargets.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ProcessorBindingTestsWithBindingTargets.java @@ -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.any()); diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ProcessorBindingTestsWithDefaults.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ProcessorBindingTestsWithDefaults.java index 7db626371..62aa951bb 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ProcessorBindingTestsWithDefaults.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ProcessorBindingTestsWithDefaults.java @@ -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.any()); diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SinkBindingTestsWithBindingTargets.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SinkBindingTestsWithBindingTargets.java index 764780972..c25f279de 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SinkBindingTestsWithBindingTargets.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SinkBindingTestsWithBindingTargets.java @@ -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.any()); diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SinkBindingTestsWithDefaults.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SinkBindingTestsWithDefaults.java index 6acf3d93f..0c9ec505a 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SinkBindingTestsWithDefaults.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SinkBindingTestsWithDefaults.java @@ -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.any()); diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SourceBindingTestsWithBindingTargets.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SourceBindingTestsWithBindingTargets.java index 08df2a56b..d6eb70581 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SourceBindingTestsWithBindingTargets.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SourceBindingTestsWithBindingTargets.java @@ -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.any()); diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SourceBindingTestsWithDefaults.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SourceBindingTestsWithDefaults.java index 728ecdd3e..1171ca2a2 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SourceBindingTestsWithDefaults.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SourceBindingTestsWithDefaults.java @@ -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.any()); diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/ChannelBindingAdapterConfigurationTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/ChannelBindingAdapterConfigurationTests.java deleted file mode 100644 index d07663db4..000000000 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/ChannelBindingAdapterConfigurationTests.java +++ /dev/null @@ -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 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 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 channels = this.adapter.getChannelsMetadata().getOutputChannels(); - List remoteBindingNames = getRemoteBindingNames(channels); - assertEquals(2, channels.size()); - assertTrue(remoteBindingNames.contains("group.0")); - assertTrue(remoteBindingNames.contains("topic:group.0")); - } - - private List getRemoteBindingNames(Collection channels) { - List list = new ArrayList(); - 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 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 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(); - } - } - -} diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/interceptor/BoundChannelsInterceptedTest.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/interceptor/BoundChannelsInterceptedTest.java index 2f833272e..959dbdc22 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/interceptor/BoundChannelsInterceptedTest.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/interceptor/BoundChannelsInterceptedTest.java @@ -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; diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/utils/MockBinderConfiguration.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/utils/MockBinderConfiguration.java index c50d54651..ef1eca15e 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/utils/MockBinderConfiguration.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/utils/MockBinderConfiguration.java @@ -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,