Switch to spring-cloud-streams naming

2015-07-08 17:02:44 +01:00
parent 71e53321dc
commit 997356cbe5
32 changed files with 34 additions and 34 deletions
--- a/spring-cloud-streams/pom.xml
+++ b/spring-cloud-streams/pom.xml
@@ -0,0 +1,67 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+	<modelVersion>4.0.0</modelVersion>
+
+	<groupId>org.springframework.cloud</groupId>
+	<artifactId>spring-cloud-streams</artifactId>
+	<version>1.0.0.BUILD-SNAPSHOT</version>
+	<packaging>jar</packaging>
+
+	<name>spring-cloud-streams</name>
+	<description>Messaging Microservices with Spring Integration</description>
+
+	<parent>
+		<groupId>org.springframework.cloud</groupId>
+		<artifactId>spring-cloud-streams-parent</artifactId>
+		<version>1.0.0.BUILD-SNAPSHOT</version>
+	</parent>
+
+	<properties>
+		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+		<java.version>1.8</java.version>
+	</properties>
+
+	<dependencies>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-actuator</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-starter</artifactId>
+			<optional>true</optional>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-web</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.xd</groupId>
+			<artifactId>spring-xd-dirt</artifactId>
+		</dependency>
+
+		<dependency>
+			<groupId>org.springframework.xd</groupId>
+			<artifactId>spring-xd-messagebus-redis</artifactId>
+			<optional>true</optional>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.xd</groupId>
+			<artifactId>spring-xd-messagebus-rabbit</artifactId>
+			<optional>true</optional>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-configuration-processor</artifactId>
+			<optional>true</optional>
+		</dependency>
+
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-test</artifactId>
+			<scope>test</scope>
+		</dependency>
+	</dependencies>
+
+</project>
--- a/spring-cloud-streams/src/main/java/org/springframework/bus/runner/EnableMessageBus.java
+++ b/spring-cloud-streams/src/main/java/org/springframework/bus/runner/EnableMessageBus.java
@@ -0,0 +1,46 @@
+/*
+ * 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.bus.runner;
+
+import java.lang.annotation.Documented;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Inherited;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+import org.springframework.bus.runner.config.LifecycleConfiguration;
+import org.springframework.bus.runner.config.MessageBusAdapterConfiguration;
+import org.springframework.bus.runner.config.RabbitServiceConfiguration;
+import org.springframework.bus.runner.config.RedisServiceConfiguration;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
+
+/**
+ * @author Dave Syer
+ *
+ */
+@Target(ElementType.TYPE)
+@Retention(RetentionPolicy.RUNTIME)
+@Documented
+@Inherited
+@Configuration
+@Import({ RedisServiceConfiguration.class, RabbitServiceConfiguration.class,
+		MessageBusAdapterConfiguration.class, LifecycleConfiguration.class })
+public @interface EnableMessageBus {
+
+}
--- a/spring-cloud-streams/src/main/java/org/springframework/bus/runner/adapter/ChannelBinding.java
+++ b/spring-cloud-streams/src/main/java/org/springframework/bus/runner/adapter/ChannelBinding.java
@@ -0,0 +1,50 @@
+/*
+ * 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.bus.runner.adapter;
+
+/**
+ * Represents a binding between a local and remote message channel.
+ *
+ * @author Dave Syer
+ * @author Mark Fisher
+ */
+public abstract class ChannelBinding {
+
+	private String localName;
+	private String remoteName;
+
+	protected ChannelBinding() {
+		this(null);
+	}
+
+	protected ChannelBinding(String localName) {
+		this.localName = localName;
+	}
+
+	public String getLocalName() {
+		return this.localName;
+	}
+
+	public String getRemoteName() {
+		return this.remoteName;
+	}
+
+	public void setRemoteName(String name) {
+		this.remoteName = name;
+	}
+
+}
--- a/spring-cloud-streams/src/main/java/org/springframework/bus/runner/adapter/ChannelLocator.java
+++ b/spring-cloud-streams/src/main/java/org/springframework/bus/runner/adapter/ChannelLocator.java
@@ -0,0 +1,28 @@
+/*
+ * 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.bus.runner.adapter;
+
+/**
+ * @author Dave Syer
+ *
+ */
+// TODO: Use DestinationResolver?
+public interface ChannelLocator {
+	
+	String locate(String name);
+
+}
--- a/spring-cloud-streams/src/main/java/org/springframework/bus/runner/adapter/ChannelsMetadata.java
+++ b/spring-cloud-streams/src/main/java/org/springframework/bus/runner/adapter/ChannelsMetadata.java
@@ -0,0 +1,57 @@
+/*
+ * 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.bus.runner.adapter;
+
+import java.util.Collection;
+import java.util.Collections;
+
+import org.springframework.bus.runner.config.MessageBusProperties;
+
+/**
+ * @author Dave Syer
+ */
+public class ChannelsMetadata {
+
+	private Collection<OutputChannelBinding> outputChannels = Collections.emptySet();
+	private Collection<InputChannelBinding> inputChannels = Collections.emptySet();
+	private MessageBusProperties module;
+
+	public MessageBusProperties getModule() {
+		return this.module;
+	}
+
+	public void setModule(MessageBusProperties module) {
+		this.module = module;
+	}
+
+	public Collection<OutputChannelBinding> getOutputChannels() {
+		return this.outputChannels;
+	}
+
+	public void setOutputChannels(Collection<OutputChannelBinding> outputChannels) {
+		this.outputChannels = outputChannels;
+	}
+
+	public Collection<InputChannelBinding> getInputChannels() {
+		return this.inputChannels;
+	}
+
+	public void setInputChannels(Collection<InputChannelBinding> inputChannels) {
+		this.inputChannels = inputChannels;
+	}
+
+}
--- a/spring-cloud-streams/src/main/java/org/springframework/bus/runner/adapter/DefaultChannelLocator.java
+++ b/spring-cloud-streams/src/main/java/org/springframework/bus/runner/adapter/DefaultChannelLocator.java
@@ -0,0 +1,82 @@
+/*
+ * 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.bus.runner.adapter;
+
+import org.springframework.bus.runner.config.MessageBusProperties;
+import org.springframework.util.StringUtils;
+
+/**
+ * @author Dave Syer
+ */
+public class DefaultChannelLocator implements ChannelLocator {
+
+	private MessageBusProperties module;
+
+	public DefaultChannelLocator(MessageBusProperties module) {
+		this.module = module;
+	}
+
+	@Override
+	public String locate(String name) {
+		String channelName = extractChannelName("input", name, this.module.getInputChannelName());
+		if (channelName!=null) {
+			return channelName;
+		}
+		channelName = extractChannelName("output", name, this.module.getOutputChannelName());
+		if (channelName!=null) {
+			return channelName;
+		}
+		return null;
+	}
+
+
+	private String extractChannelName(String start, String name, String externalChannelName) {
+		if (name.equals(start)) {
+			return externalChannelName;
+		}
+		else if (name.startsWith(start + ".") || name.startsWith(start + "_")) {
+			String prefix = "";
+			String channelName = name.substring(start.length() + 1);
+			if (channelName.contains(":")) {
+				String[] tokens = channelName.split(":", 2);
+				String type = tokens[0];
+				if ("queue".equals(type)) {
+					// omit the type for a queue
+					if (StringUtils.hasText(tokens[1])) {
+						prefix = tokens[1] + ".";
+					}
+				}
+				else {
+					prefix = channelName + (channelName.endsWith(":") ? "" : ".");
+				}
+			}
+			else {
+				prefix = channelName + ".";
+			}
+			return prefix + getPlainChannelName(externalChannelName);
+		}
+		return null;
+	}
+
+	private String getPlainChannelName(String name) {
+		if (name.contains(":")) {
+			name = name.substring(name.indexOf(":") + 1);
+		}
+		return name;
+	}
+
+}
--- a/spring-cloud-streams/src/main/java/org/springframework/bus/runner/adapter/Input.java
+++ b/spring-cloud-streams/src/main/java/org/springframework/bus/runner/adapter/Input.java
@@ -0,0 +1,41 @@
+/*
+ * 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.bus.runner.adapter;
+
+import java.lang.annotation.Documented;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Inherited;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+import org.springframework.beans.factory.annotation.Qualifier;
+
+/**
+ * Qualifier annotation for a bean relating input channels.
+ *
+ * @author Dave Syer
+ */
+@Qualifier
+@Target({ ElementType.FIELD, ElementType.METHOD, ElementType.PARAMETER, ElementType.TYPE,
+	ElementType.ANNOTATION_TYPE })
+@Retention(RetentionPolicy.RUNTIME)
+@Inherited
+@Documented
+public @interface Input {
+
+}
--- a/spring-cloud-streams/src/main/java/org/springframework/bus/runner/adapter/InputChannelBinding.java
+++ b/spring-cloud-streams/src/main/java/org/springframework/bus/runner/adapter/InputChannelBinding.java
@@ -0,0 +1,32 @@
+/*
+ * 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.bus.runner.adapter;
+
+/**
+ * @author Dave Syer
+ */
+public class InputChannelBinding extends ChannelBinding {
+
+	protected InputChannelBinding() {
+		super(null);
+	}
+
+	public InputChannelBinding(String localName) {
+		super(localName);
+	}
+
+}
--- a/spring-cloud-streams/src/main/java/org/springframework/bus/runner/adapter/MessageBusAdapter.java
+++ b/spring-cloud-streams/src/main/java/org/springframework/bus/runner/adapter/MessageBusAdapter.java
@@ -0,0 +1,433 @@
+/*
+ * 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.bus.runner.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.bus.runner.config.MessageBusProperties;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.ApplicationContextAware;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.context.Lifecycle;
+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.integration.support.channel.BeanFactoryChannelResolver;
+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;
+import org.springframework.xd.dirt.integration.bus.MessageBus;
+import org.springframework.xd.dirt.integration.bus.XdHeaders;
+
+/**
+ * @author Mark Fisher
+ * @author Dave Syer
+ */
+@ManagedResource
+public class MessageBusAdapter implements Lifecycle, ApplicationContextAware {
+
+	private static Logger logger = LoggerFactory.getLogger(MessageBusAdapter.class);
+
+	private MessageBus messageBus;
+	private MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory();
+
+	private Collection<OutputChannelBinding> outputChannels = Collections.emptySet();
+	private Collection<InputChannelBinding> inputChannels = Collections.emptySet();
+
+	private boolean running = false;
+
+	private final AtomicBoolean active = new AtomicBoolean(false);
+
+	private boolean trackHistory = false;
+
+	private MessageBusProperties module;
+
+	private ConfigurableApplicationContext applicationContext;
+
+	private ChannelLocator inputChannelLocator;
+
+	private ChannelLocator outputChannelLocator;
+
+	private DestinationResolver<MessageChannel> channelResolver;
+
+	private Map<String, String> bindings = new HashMap<String, String>();
+
+	public MessageBusAdapter(MessageBusProperties module, MessageBus messageBus) {
+		this.module = module;
+		this.messageBus = messageBus;
+		this.inputChannelLocator = new DefaultChannelLocator(module);
+		this.outputChannelLocator = new DefaultChannelLocator(module);
+	}
+
+	public void setInputChannelLocator(ChannelLocator channelLocator) {
+		this.inputChannelLocator = channelLocator;
+	}
+
+	public void setOutputChannelLocator(ChannelLocator channelLocator) {
+		this.outputChannelLocator = channelLocator;
+	}
+
+	@Override
+	public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
+		this.applicationContext = (ConfigurableApplicationContext) applicationContext;
+		this.channelResolver = new BeanFactoryChannelResolver(applicationContext);
+	}
+
+	public void setChannelResolver(DestinationResolver<MessageChannel> channelResolver) {
+		this.channelResolver = channelResolver;
+	}
+
+	public void setMessageBuilderFactory(MessageBuilderFactory messageBuilderFactory) {
+		this.messageBuilderFactory = messageBuilderFactory;
+	}
+
+	public void setTrackHistory(boolean trackHistory) {
+		this.trackHistory = trackHistory;
+	}
+
+	public void setOutputChannels(Collection<OutputChannelBinding> outputChannels) {
+		this.outputChannels = new LinkedHashSet<OutputChannelBinding>(outputChannels);
+	}
+
+	public void setInputChannels(Collection<InputChannelBinding> inputChannels) {
+		this.inputChannels = new LinkedHashSet<InputChannelBinding>(inputChannels);
+	}
+
+	public ChannelsMetadata getChannelsMetadata() {
+		ChannelsMetadata channels = new ChannelsMetadata();
+		channels.setModule(this.module);
+		channels.setInputChannels(new LinkedHashSet<InputChannelBinding>(this.inputChannels));
+		channels.setOutputChannels(new LinkedHashSet<OutputChannelBinding>(this.outputChannels));
+		return channels;
+	}
+
+	public OutputChannelBinding getOutputChannel(String name) {
+		if (name == null) {
+			return null;
+		}
+		for (OutputChannelBinding binding : this.outputChannels) {
+			if (name.equals(binding.getRemoteName())) {
+				return binding;
+			}
+		}
+		for (OutputChannelBinding binding : this.outputChannels) {
+			if (name.equals(binding.getLocalName())) {
+				return binding;
+			}
+		}
+		return null;
+	}
+
+	public InputChannelBinding getInputChannel(String name) {
+		if (name == null) {
+			return null;
+		}
+		for (InputChannelBinding binding : this.inputChannels) {
+			if (name.equals(binding.getRemoteName())) {
+				return binding;
+			}
+		}
+		for (InputChannelBinding binding : this.inputChannels) {
+			if (name.equals(binding.getLocalName())) {
+				return binding;
+			}
+		}
+		return null;
+	}
+
+	public void tap(String outputChannel) {
+		OutputChannelBinding channel = getOutputChannel(outputChannel);
+		if (channel == null || channel.isTapped()) {
+			return;
+		}
+		createAndBindTapChannel(channel.getTapChannelName(), channel.getLocalName());
+		channel.setTapped(true);
+	}
+
+	public void untap(String outputChannel) {
+		OutputChannelBinding channel = getOutputChannel(outputChannel);
+		if (channel == null || !channel.isTapped()) {
+			return;
+		}
+		String tapChannelName = channel.getTapChannelName();
+		this.messageBus.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.messageBus.unbindConsumers(name);
+		}
+		for (OutputChannelBinding binding : this.outputChannels) {
+			String name = this.bindings.get(binding.getRemoteName());
+			if (name == null) {
+				continue;
+			}
+			this.messageBus.unbindProducers(name);
+			if (binding.isTapped()) {
+				String tapChannelName = binding.getTapChannelName();
+				this.messageBus.unbindProducers(tapChannelName);
+			}
+		}
+	}
+
+	protected final boolean bindChannels() {
+		if (!locateChannels()) {
+			return false;
+		}
+		Map<String, Object> historyProperties = new LinkedHashMap<String, Object>();
+		if (this.trackHistory) {
+			// TODO: addHistoryTag();
+		}
+		for (OutputChannelBinding binding : this.outputChannels) {
+			String name = binding.getRemoteName();
+			MessageChannel outputChannel = this.channelResolver.resolveDestination(binding.getLocalName());
+			bindMessageProducer(outputChannel, name, this.module.getProducerProperties());
+			if (binding.isTapped()) {
+				String tapChannelName = getTapChannelName(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.outputChannelLocator.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.inputChannelLocator.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;
+	}
+
+	// TODO: move this to ChannelLocator?
+	private String getTapChannelName(String name) {
+		return !isDefaultOuputChannel(name) ? this.module.getTapChannelName(getPlainChannelName(name))
+				: this.module.getTapChannelName();
+	}
+
+	// TODO: move this to ChannelLocator?
+	private String getPlainChannelName(String name) {
+		if (name.contains(":")) {
+			name = name.substring(name.indexOf(":") + 1);
+		}
+		return name;
+	}
+
+	// TODO: move this to ChannelLocator?
+	private boolean isDefaultOuputChannel(String channelName) {
+		if (channelName.contains(":")) {
+			String[] tokens = channelName.split(":", 2);
+			channelName = tokens[1];
+		}
+		return channelName.equals(this.module.getOutputChannelName());
+	}
+
+	/*
+	 * Following methods copied from parent to support the bindChannels() method above
+	 */
+
+	private void bindMessageConsumer(MessageChannel inputChannel,
+			String inputChannelName, Properties consumerProperties) {
+		if (isChannelPubSub(inputChannelName)) {
+			this.messageBus.bindPubSubConsumer(inputChannelName, inputChannel, consumerProperties);
+		}
+		else {
+			this.messageBus.bindConsumer(inputChannelName, inputChannel, consumerProperties);
+		}
+	}
+
+	private void bindMessageProducer(MessageChannel outputChannel,
+			String outputChannelName, Properties producerProperties) {
+		if (isChannelPubSub(outputChannelName)) {
+			this.messageBus.bindPubSubProducer(outputChannelName, outputChannel, producerProperties);
+		}
+		else {
+			this.messageBus.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 MessageBus}'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.messageBus.bindPubSubProducer(tapChannelName, tapChannel, null); // TODO
+			// tap
+			// producer
+			// props
+			tapOutputChannel(tapChannel, (ChannelInterceptorAware) channel);
+		}
+		else {
+			if (logger.isDebugEnabled()) {
+				logger.debug("output channel is not interceptor aware. Tap will not be created.");
+			}
+		}
+	}
+
+	private MessageChannel tapOutputChannel(MessageChannel tapChannel, ChannelInterceptorAware outputChannel) {
+		outputChannel.addInterceptor(new WireTap(tapChannel));
+		return tapChannel;
+	}
+
+	private void track(MessageChannel channel, final Map<String, Object> historyProps) {
+		if (channel instanceof ChannelInterceptorAware) {
+			((ChannelInterceptorAware) channel)
+			.addInterceptor(new ChannelInterceptorAdapter() {
+
+				@Override
+				public Message<?> preSend(Message<?> message, MessageChannel channel) {
+					@SuppressWarnings("unchecked")
+					Collection<Map<String, Object>> history = (Collection<Map<String, Object>>) message
+					.getHeaders().get(XdHeaders.XD_HISTORY);
+					if (history == null) {
+						history = new ArrayList<Map<String, Object>>(1);
+					}
+					else {
+						history = new ArrayList<Map<String, Object>>(history);
+					}
+					Map<String, Object> map = new LinkedHashMap<String, Object>();
+					map.putAll(historyProps);
+					map.put("thread", Thread.currentThread().getName());
+					history.add(map);
+					Message<?> out = MessageBusAdapter.this.messageBuilderFactory.fromMessage(message)
+							.setHeader(XdHeaders.XD_HISTORY, history).build();
+					return out;
+				}
+			});
+		}
+	}
+
+}
--- a/spring-cloud-streams/src/main/java/org/springframework/bus/runner/adapter/Output.java
+++ b/spring-cloud-streams/src/main/java/org/springframework/bus/runner/adapter/Output.java
@@ -0,0 +1,41 @@
+/*
+ * 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.bus.runner.adapter;
+
+import java.lang.annotation.Documented;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Inherited;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+import org.springframework.beans.factory.annotation.Qualifier;
+
+/**
+ * Qualifier annotation for a bean relating output channels.
+ *
+ * @author Dave Syer
+ */
+@Qualifier
+@Target({ ElementType.FIELD, ElementType.METHOD, ElementType.PARAMETER, ElementType.TYPE,
+	ElementType.ANNOTATION_TYPE })
+@Retention(RetentionPolicy.RUNTIME)
+@Inherited
+@Documented
+public @interface Output {
+
+}
--- a/spring-cloud-streams/src/main/java/org/springframework/bus/runner/adapter/OutputChannelBinding.java
+++ b/spring-cloud-streams/src/main/java/org/springframework/bus/runner/adapter/OutputChannelBinding.java
@@ -0,0 +1,52 @@
+/*
+ * 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.bus.runner.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;
+	}
+
+}
--- a/spring-cloud-streams/src/main/java/org/springframework/bus/runner/adapter/discovery/DiscoveryClientAutoConfiguration.java
+++ b/spring-cloud-streams/src/main/java/org/springframework/bus/runner/adapter/discovery/DiscoveryClientAutoConfiguration.java
@@ -0,0 +1,74 @@
+/*
+ * 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.bus.runner.adapter.discovery;
+
+import javax.annotation.PostConstruct;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.bus.runner.adapter.ChannelLocator;
+import org.springframework.bus.runner.adapter.Output;
+import org.springframework.bus.runner.adapter.MessageBusAdapter;
+import org.springframework.bus.runner.adapter.Input;
+import org.springframework.cloud.client.discovery.DiscoveryClient;
+import org.springframework.cloud.client.discovery.event.HeartbeatEvent;
+import org.springframework.cloud.client.discovery.event.HeartbeatMonitor;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.event.EventListener;
+
+/**
+ * Autoconfiguration for use when user has provided a
+ * {@link DiscoveryClientChannelLocator}. Listens for changes in the service registry and
+ * rebinds the external channels as needed.
+ *
+ * @author Dave Syer
+ */
+@Configuration
+@ConditionalOnClass(DiscoveryClient.class)
+public class DiscoveryClientAutoConfiguration {
+
+	private HeartbeatMonitor monitor = new HeartbeatMonitor();
+
+	@Autowired
+	private MessageBusAdapter adapter;
+
+	@Autowired(required = false)
+	@Input
+	private ChannelLocator inputChannelLocator;
+
+	@Autowired(required = false)
+	@Output
+	private ChannelLocator outputChannelLocator;
+
+	private boolean enabled = false;
+
+	@PostConstruct
+	public void init() {
+		if (this.inputChannelLocator instanceof DiscoveryClientChannelLocator
+				|| this.outputChannelLocator instanceof DiscoveryClientChannelLocator) {
+			this.enabled = true;
+		}
+	}
+
+	@EventListener
+	public void discoveryHeartbeat(HeartbeatEvent event) {
+		if (this.enabled && this.monitor.update(event.getValue())) {
+			this.adapter.rebind();
+		}
+	}
+
+}
--- a/spring-cloud-streams/src/main/java/org/springframework/bus/runner/adapter/discovery/DiscoveryClientChannelLocator.java
+++ b/spring-cloud-streams/src/main/java/org/springframework/bus/runner/adapter/discovery/DiscoveryClientChannelLocator.java
@@ -0,0 +1,99 @@
+/*
+ * 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.bus.runner.adapter.discovery;
+
+import java.net.URI;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Random;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.springframework.bus.runner.adapter.ChannelBinding;
+import org.springframework.bus.runner.adapter.ChannelLocator;
+import org.springframework.bus.runner.adapter.ChannelsMetadata;
+import org.springframework.cloud.client.ServiceInstance;
+import org.springframework.cloud.client.discovery.DiscoveryClient;
+import org.springframework.web.client.RestOperations;
+import org.springframework.web.client.RestTemplate;
+import org.springframework.web.util.UriComponentsBuilder;
+
+/**
+ * @author Dave Syer
+ */
+public class DiscoveryClientChannelLocator implements ChannelLocator {
+
+	private Log logger = LogFactory.getLog(DiscoveryClientChannelLocator.class);
+
+	private DiscoveryClient discovery;
+
+	private RestOperations restTemplate = new RestTemplate();
+
+	private String serviceId;
+
+	public DiscoveryClientChannelLocator(DiscoveryClient discovery, String serviceId) {
+		this.discovery = discovery;
+		this.serviceId = serviceId;
+	}
+
+	public void setRestTemplate(RestOperations restTemplate) {
+		this.restTemplate = restTemplate;
+	}
+
+	@Override
+	public String locate(String name) {
+		List<ServiceInstance> instances = this.discovery.getInstances(this.serviceId);
+		if (instances == null || instances.isEmpty()) {
+			return null;
+		}
+		URI uri = pickUrl(instances);
+		try {
+			ChannelsMetadata channels = this.restTemplate.getForObject(uri, ChannelsMetadata.class);
+			Collection<? extends ChannelBinding> bindings = Collections.emptySet();
+			if (name.startsWith("input")) {
+				name = name.replace("input", "output");
+				bindings = channels.getOutputChannels();
+			}
+			else if (name.startsWith("output")) {
+				name = name.replace("output", "input");
+				bindings = channels.getInputChannels();
+			}
+			for (ChannelBinding binding : bindings) {
+				if (name.equals(binding.getLocalName())) {
+					this.logger.debug("Discovered channel for '" + this.serviceId + "' ("
+							+ name + "=" + binding.getRemoteName() + ")");
+					return binding.getRemoteName();
+				}
+			}
+		}
+		catch (Exception e) {
+			this.logger.warn("Could not discover channel for '" + this.serviceId + "' ("
+					+ e.getClass() + ": " + e.getMessage() + ")");
+			return null;
+		}
+		this.logger.warn("No channel disccovered for '" + this.serviceId + "' (" + name + ")");
+		return null;
+	}
+
+	private URI pickUrl(List<ServiceInstance> instances) {
+		return UriComponentsBuilder
+				.fromUri(instances.get(new Random().nextInt(instances.size())).getUri())
+				.path("channels").build().toUri();
+	}
+
+}
--- a/spring-cloud-streams/src/main/java/org/springframework/bus/runner/config/LifecycleConfiguration.java
+++ b/spring-cloud-streams/src/main/java/org/springframework/bus/runner/config/LifecycleConfiguration.java
@@ -0,0 +1,44 @@
+/*
+ * 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.bus.runner.config;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.CommandLineRunner;
+import org.springframework.bus.runner.adapter.MessageBusAdapter;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * @author Dave Syer
+ *
+ */
+@Configuration
+public class LifecycleConfiguration implements CommandLineRunner {
+
+	@Autowired
+	private MessageBusProperties module;
+
+	@Autowired
+	private MessageBusAdapter adapter;
+
+	@Override
+	public void run(String... args) throws Exception {
+		if (!adapter.isRunning() && module.isAutoStartup()) {
+			adapter.start();
+		}
+	}
+
+}
--- a/spring-cloud-streams/src/main/java/org/springframework/bus/runner/config/MessageBusAdapterConfiguration.java
+++ b/spring-cloud-streams/src/main/java/org/springframework/bus/runner/config/MessageBusAdapterConfiguration.java
@@ -0,0 +1,161 @@
+/*
+ * 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.bus.runner.config;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.LinkedHashSet;
+import java.util.Properties;
+import java.util.Set;
+
+import org.aopalliance.intercept.MethodInterceptor;
+import org.aopalliance.intercept.MethodInvocation;
+import org.springframework.aop.framework.ProxyFactory;
+import org.springframework.aop.target.LazyInitTargetSource;
+import org.springframework.beans.factory.BeanFactoryUtils;
+import org.springframework.beans.factory.ListableBeanFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.bus.runner.adapter.ChannelLocator;
+import org.springframework.bus.runner.adapter.Input;
+import org.springframework.bus.runner.adapter.InputChannelBinding;
+import org.springframework.bus.runner.adapter.MessageBusAdapter;
+import org.springframework.bus.runner.adapter.Output;
+import org.springframework.bus.runner.adapter.OutputChannelBinding;
+import org.springframework.bus.runner.endpoint.ChannelsEndpoint;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.ImportResource;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.util.Assert;
+import org.springframework.xd.dirt.integration.bus.MessageBus;
+import org.springframework.xd.dirt.integration.bus.MessageBusAwareRouterBeanPostProcessor;
+
+/**
+ * @author Dave Syer
+ *
+ */
+@Configuration
+@ImportResource("classpath*:/META-INF/spring-xd/bus/codec.xml")
+@EnableConfigurationProperties(MessageBusProperties.class)
+public class MessageBusAdapterConfiguration {
+
+	@Autowired
+	private MessageBusProperties module;
+
+	@Autowired
+	private ListableBeanFactory beanFactory;
+
+	@Autowired(required=false)
+	@Input
+	private ChannelLocator inputChannelLocator;
+
+	@Autowired(required=false)
+	@Output
+	private ChannelLocator outputChannelLocator;
+
+	@Bean
+	public MessageBusAdapter messageBusAdapter(MessageBusProperties module,
+			MessageBus messageBus) {
+		MessageBusAdapter adapter = new MessageBusAdapter(module, messageBus);
+		adapter.setOutputChannels(getOutputChannels());
+		adapter.setInputChannels(getInputChannels());
+		if (this.inputChannelLocator!=null) {
+			adapter.setInputChannelLocator(this.inputChannelLocator);
+		}
+		if (this.outputChannelLocator!=null) {
+			adapter.setOutputChannelLocator(this.outputChannelLocator);
+		}
+		return adapter;
+	}
+
+	@Bean
+	public ChannelsEndpoint channelsEndpoint(MessageBusAdapter adapter) {
+		return new ChannelsEndpoint(adapter);
+	}
+
+	protected Collection<OutputChannelBinding> getOutputChannels() {
+		Set<OutputChannelBinding> channels = new LinkedHashSet<OutputChannelBinding>();
+		String[] names = BeanFactoryUtils.beanNamesForTypeIncludingAncestors(
+				this.beanFactory, MessageChannel.class);
+		for (String name : names) {
+			if (name.startsWith("output")) {
+				channels.add(new OutputChannelBinding(name));
+			}
+		}
+		return channels;
+	}
+
+	protected Collection<InputChannelBinding> getInputChannels() {
+		Set<InputChannelBinding> channels = new LinkedHashSet<InputChannelBinding>();
+		String[] names = BeanFactoryUtils.beanNamesForTypeIncludingAncestors(
+				this.beanFactory, MessageChannel.class);
+		for (String name : names) {
+			if (name.startsWith("input")) {
+				channels.add(new InputChannelBinding(name));
+			}
+		}
+		return channels;
+	}
+
+	// Nested class to avoid instantiating all of the above early
+	@Configuration
+	protected static class MessageBusAwareRouterConfiguration {
+
+		@Autowired
+		private ListableBeanFactory beanFactory;
+
+		@Bean
+		public MessageBusAwareRouterBeanPostProcessor messageBusAwareRouterBeanPostProcessor() {
+
+			return new MessageBusAwareRouterBeanPostProcessor(createLazyProxy(
+					this.beanFactory, MessageBus.class), new Properties());
+		}
+
+		private <T> T createLazyProxy(ListableBeanFactory beanFactory, Class<T> type) {
+			ProxyFactory factory = new ProxyFactory();
+			LazyInitTargetSource source = new LazyInitTargetSource();
+			source.setTargetClass(type);
+			source.setTargetBeanName(getBeanNameFor(beanFactory, MessageBus.class));
+			source.setBeanFactory(beanFactory);
+			factory.setTargetSource(source);
+			factory.addAdvice(new PassthruAdvice());
+			factory.setInterfaces(new Class<?>[] { type });
+			@SuppressWarnings("unchecked")
+			T proxy = (T) factory.getProxy();
+			return proxy;
+		}
+
+		private String getBeanNameFor(ListableBeanFactory beanFactory, Class<?> type) {
+			String[] names = BeanFactoryUtils.beanNamesForTypeIncludingAncestors(
+					beanFactory, type, false, false);
+			Assert.state(names.length == 1, "No unique MessageBus (found " + names.length
+					+ ": " + Arrays.asList(names) + ")");
+			return names[0];
+		}
+
+		private class PassthruAdvice implements MethodInterceptor {
+
+			@Override
+			public Object invoke(MethodInvocation invocation) throws Throwable {
+				return invocation.proceed();
+			}
+
+		}
+
+	}
+
+}
--- a/spring-cloud-streams/src/main/java/org/springframework/bus/runner/config/MessageBusProperties.java
+++ b/spring-cloud-streams/src/main/java/org/springframework/bus/runner/config/MessageBusProperties.java
@@ -0,0 +1,236 @@
+/*
+ * 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.bus.runner.config;
+
+import java.util.Properties;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.util.Assert;
+import org.springframework.xd.dirt.integration.bus.BusUtils;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonInclude.Include;
+
+/**
+ * @author Dave Syer
+ *
+ */
+@ConfigurationProperties("spring.bus")
+@JsonInclude(Include.NON_DEFAULT)
+public class MessageBusProperties {
+
+	private String name = "module";
+
+	private String group = "group";
+
+	private int index = 0;
+
+	private String outputChannelName;
+
+	private String inputChannelName;
+
+	private String type = "processor";
+
+	private Properties consumerProperties = new Properties();
+
+	private Properties producerProperties = new Properties();
+
+	private Tap tap;
+
+	private Discovery discovery = new Discovery();
+
+	private boolean autoStartup = true;
+
+	public String getName() {
+		return name;
+	}
+
+	public void setName(String name) {
+		this.name = name;
+	}
+
+	public String getGroup() {
+		return group;
+	}
+
+	public void setGroup(String group) {
+		this.group = group;
+	}
+
+	public int getIndex() {
+		return index;
+	}
+
+	public void setIndex(int index) {
+		this.index = index;
+	}
+
+	public String getInputChannelName() {
+		if (isTap()) {
+			return String.format("%s.%s.%s", BusUtils.constructTapPrefix(tap.getGroup()),
+					tap.getName(), tap.getIndex());
+		}
+		return (inputChannelName != null) ? inputChannelName : BusUtils
+				.constructPipeName(group, index > 0 ? index - 1 : index);
+	}
+
+	public String getOutputChannelName() {
+		return (outputChannelName != null) ? outputChannelName : BusUtils
+				.constructPipeName(group, index);
+	}
+
+	public String getTapChannelName() {
+		return getTapChannelName(group);
+	}
+
+	public String getTapChannelName(String prefix) {
+		Assert.isTrue(!type.equals("job"), "Job module type not supported.");
+		// for Stream return channel name with indexed elements
+		return String
+				.format("%s.%s.%s", BusUtils.constructTapPrefix(prefix), name, index);
+	}
+
+	public void setOutputChannelName(String outputChannelName) {
+		this.outputChannelName = outputChannelName;
+	}
+
+	public void setInputChannelName(String inputChannelName) {
+		this.inputChannelName = inputChannelName;
+	}
+
+	public String getType() {
+		return type;
+	}
+
+	public void setType(String type) {
+		this.type = type;
+	}
+
+	public Properties getConsumerProperties() {
+		return consumerProperties;
+	}
+
+	public void setConsumerProperties(Properties consumerProperties) {
+		this.consumerProperties = consumerProperties;
+	}
+
+	public Properties getProducerProperties() {
+		return producerProperties;
+	}
+
+	public void setProducerProperties(Properties producerProperties) {
+		this.producerProperties = producerProperties;
+	}
+
+	public boolean isAutoStartup() {
+		return autoStartup;
+	}
+
+	public void setAutoStartup(boolean autoStartup) {
+		this.autoStartup = autoStartup;
+	}
+
+	public Tap getTap() {
+		return tap;
+	}
+
+	public void setTap(Tap tap) {
+		this.tap = tap;
+	}
+
+	private boolean isTap() {
+		if (tap != null) {
+			Assert.state(tap.getName() != null, "Tap name not provided");
+			Assert.state(!tap.getGroup().equals(group),
+					"Tap group cannot be the same as module group");
+		}
+		return tap != null;
+	}
+
+	public Discovery getDiscovery() {
+		return discovery;
+	}
+
+	public static class Discovery {
+
+		private boolean enabled = false;
+
+		private String inputServiceId;
+
+		private String outputServiceId;
+
+		public boolean isEnabled() {
+			return enabled;
+		}
+
+		public void setEnabled(boolean enabled) {
+			this.enabled = enabled;
+		}
+
+		public String getInputServiceId() {
+			return inputServiceId;
+		}
+
+		public void setInputServiceId(String inputServiceId) {
+			this.inputServiceId = inputServiceId;
+		}
+
+		public String getOutputServiceId() {
+			return outputServiceId;
+		}
+
+		public void setOutputServiceId(String outputServiceId) {
+			this.outputServiceId = outputServiceId;
+		}
+
+	}
+
+	public static class Tap {
+
+		private String group = "group";
+
+		private String name;
+
+		public String getName() {
+			return name;
+		}
+
+		public void setName(String name) {
+			this.name = name;
+		}
+
+		private int index = 0;
+
+		public String getGroup() {
+			return group;
+		}
+
+		public void setGroup(String group) {
+			this.group = group;
+		}
+
+		public int getIndex() {
+			return index;
+		}
+
+		public void setIndex(int index) {
+			this.index = index;
+		}
+
+	}
+
+}
--- a/spring-cloud-streams/src/main/java/org/springframework/bus/runner/config/RabbitServiceConfiguration.java
+++ b/spring-cloud-streams/src/main/java/org/springframework/bus/runner/config/RabbitServiceConfiguration.java
@@ -0,0 +1,54 @@
+/*
+ * 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.bus.runner.config;
+
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.cloud.Cloud;
+import org.springframework.cloud.CloudFactory;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.ImportResource;
+import org.springframework.context.annotation.Profile;
+import org.springframework.context.annotation.PropertySource;
+import org.springframework.xd.dirt.integration.rabbit.RabbitMessageBus;
+
+/**
+ * Bind to services, either locally or in a Lattice environment.
+ *
+ * @author Mark Fisher
+ * @author Dave Syer
+ */
+@Configuration
+@ConditionalOnClass(RabbitMessageBus.class)
+@ImportResource({ "classpath*:/META-INF/spring-xd/bus/rabbit-bus.xml",
+		"classpath*:/META-INF/spring-xd/analytics/rabbit-analytics.xml" })
+@PropertySource("classpath:/META-INF/spring-bus/rabbit-bus.properties")
+public class RabbitServiceConfiguration {
+	@Configuration
+	@Profile("cloud")
+	protected static class CloudConfig {
+		@Bean
+		public Cloud cloud() {
+		  return new CloudFactory().getCloud();
+		}
+		@Bean
+		ConnectionFactory redisConnectionFactory(Cloud cloud) {
+			return cloud.getSingletonServiceConnector(ConnectionFactory.class, null);
+		}
+	}
+}
--- a/spring-cloud-streams/src/main/java/org/springframework/bus/runner/config/RedisServiceConfiguration.java
+++ b/spring-cloud-streams/src/main/java/org/springframework/bus/runner/config/RedisServiceConfiguration.java
@@ -0,0 +1,56 @@
+/*
+ * 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.bus.runner.config;
+
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.cloud.Cloud;
+import org.springframework.cloud.CloudFactory;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.ImportResource;
+import org.springframework.context.annotation.Profile;
+import org.springframework.context.annotation.PropertySource;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.xd.dirt.integration.redis.RedisMessageBus;
+
+/**
+ * Bind to services, either locally or in a Lattice environment.
+ *
+ * @author Mark Fisher
+ * @author Dave Syer
+ */
+@Configuration
+@ConditionalOnClass(RedisMessageBus.class)
+@ImportResource({ "classpath*:/META-INF/spring-xd/bus/redis-bus.xml",
+		"classpath*:/META-INF/spring-xd/analytics/redis-analytics.xml" })
+@PropertySource("classpath:/META-INF/spring-bus/redis-bus.properties")
+public class RedisServiceConfiguration {
+
+	@Configuration
+	@Profile("cloud")
+	protected static class CloudConfig {
+		@Bean
+		public Cloud cloud() {
+		  return new CloudFactory().getCloud();
+		}
+		@Bean
+		RedisConnectionFactory redisConnectionFactory(Cloud cloud) {
+			return cloud.getSingletonServiceConnector(RedisConnectionFactory.class, null);
+		}
+	}
+
+}
--- a/spring-cloud-streams/src/main/java/org/springframework/bus/runner/endpoint/ChannelsEndpoint.java
+++ b/spring-cloud-streams/src/main/java/org/springframework/bus/runner/endpoint/ChannelsEndpoint.java
@@ -0,0 +1,79 @@
+/*
+ * 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.bus.runner.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.bus.runner.adapter.ChannelsMetadata;
+import org.springframework.bus.runner.adapter.MessageBusAdapter;
+import org.springframework.bus.runner.adapter.OutputChannelBinding;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestMethod;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+
+/**
+ * @author Dave Syer
+ */
+@RestController
+public class ChannelsEndpoint extends AbstractEndpoint<Map<String, ?>> {
+	
+	private MessageBusAdapter adapter;
+
+	public ChannelsEndpoint(MessageBusAdapter adapter) {
+		super("channels");
+		this.adapter = adapter;
+	}
+
+	@RequestMapping(value="/channels/taps")
+	public List<OutputChannelBinding> taps() {
+		List<OutputChannelBinding> list = new ArrayList<OutputChannelBinding>();
+		for (OutputChannelBinding binding : adapter.getChannelsMetadata().getOutputChannels()) {
+			if (binding.isTapped()) {
+				list.add(binding);
+			}
+		}
+		return list ;
+	}
+
+	@RequestMapping(value="/channels/taps", method=RequestMethod.POST)
+	public OutputChannelBinding tap(@RequestParam String channel) {
+		adapter.tap(channel);
+		return adapter.getOutputChannel(channel);
+	}
+
+	@RequestMapping(value="/channels/taps", method=RequestMethod.DELETE)
+	public OutputChannelBinding untap(@RequestParam String channel) {
+		adapter.untap(channel);
+		return adapter.getOutputChannel(channel);
+	}
+
+	@Override
+	public Map<String, ?> invoke() {
+		LinkedHashMap<String, Object> map = new LinkedHashMap<String, Object>();
+		ChannelsMetadata channels = adapter.getChannelsMetadata();
+		map.put("inputChannels", channels.getInputChannels());
+		map.put("outputChannels", channels.getOutputChannels());
+		map.put("module", channels.getModule());
+		return map;
+	}
+
+}
--- a/spring-cloud-streams/src/main/resources/META-INF/spring-bus/rabbit-bus.properties
+++ b/spring-cloud-streams/src/main/resources/META-INF/spring-bus/rabbit-bus.properties
@@ -0,0 +1,23 @@
+xd.messagebus.rabbit.default.ackMode:                   AUTO
+xd.messagebus.rabbit.default.autoBindDLQ:               false
+xd.messagebus.rabbit.default.backOffInitialInterval:    1000
+xd.messagebus.rabbit.default.backOffMaxInterval:        10000
+xd.messagebus.rabbit.default.backOffMultiplier:         2.0
+xd.messagebus.rabbit.default.batchBufferLimit:          10000
+xd.messagebus.rabbit.default.batchingEnabled:   false
+xd.messagebus.rabbit.default.batchSize: 100
+xd.messagebus.rabbit.default.batchTimeout:      5000
+xd.messagebus.rabbit.default.compress:  false
+xd.messagebus.rabbit.default.concurrency:       1
+xd.messagebus.rabbit.default.deliveryMode:      PERSISTENT
+xd.messagebus.rabbit.default.durableSubscription:       false
+xd.messagebus.rabbit.default.maxAttempts:       3
+xd.messagebus.rabbit.default.maxConcurrency:    1
+xd.messagebus.rabbit.default.prefix:    xdbus.
+xd.messagebus.rabbit.default.prefetch:  1
+xd.messagebus.rabbit.default.replyHeaderPatterns:       STANDARD_REPLY_HEADERS,*
+xd.messagebus.rabbit.default.republishToDLQ:    false
+xd.messagebus.rabbit.default.requestHeaderPatterns:     STANDARD_REQUEST_HEADERS,*
+xd.messagebus.rabbit.default.requeue:   true
+xd.messagebus.rabbit.default.transacted:false
+xd.messagebus.rabbit.default.txSize:    1
--- a/spring-cloud-streams/src/main/resources/META-INF/spring-bus/redis-bus.properties
+++ b/spring-cloud-streams/src/main/resources/META-INF/spring-bus/redis-bus.properties
@@ -0,0 +1,5 @@
+xd.messagebus.redis.default.backOffInitialInterval:    1000
+xd.messagebus.redis.default.backOffMaxInterval:        10000
+xd.messagebus.redis.default.backOffMultiplier:         2.0
+xd.messagebus.redis.default.concurrency:               1
+xd.messagebus.redis.default.maxAttempts:               3
--- a/spring-cloud-streams/src/main/resources/META-INF/spring.factories
+++ b/spring-cloud-streams/src/main/resources/META-INF/spring.factories
@@ -0,0 +1,3 @@
+# AutoConfiguration
+org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
+org.springframework.bus.runner.adapter.discovery.DiscoveryClientAutoConfiguration
--- a/spring-cloud-streams/src/test/java/org/springframework/bus/runner/adapter/DefaultChannelLocatorTests.java
+++ b/spring-cloud-streams/src/test/java/org/springframework/bus/runner/adapter/DefaultChannelLocatorTests.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.bus.runner.adapter;
+
+import static org.junit.Assert.assertEquals;
+
+import org.junit.Test;
+import org.springframework.bus.runner.config.MessageBusProperties;
+
+/**
+ * @author Dave Syer
+ *
+ */
+public class DefaultChannelLocatorTests {
+
+	private MessageBusProperties module = new MessageBusProperties();
+
+	private DefaultChannelLocator locator = new DefaultChannelLocator(this.module);
+
+	@Test
+	public void oneOutput() throws Exception {
+		assertEquals("group.0", this.locator.locate("output"));
+	}
+
+	@Test
+	public void oneOutputTopic() throws Exception {
+		assertEquals("topic:group.0", this.locator.locate("output.topic:"));
+	}
+
+	@Test
+	public void outputWithNamedTopic() throws Exception {
+		assertEquals("topic:foo.group.0", this.locator.locate("output.topic:foo"));
+	}
+
+	@Test
+	public void outputWithNamedQueue() throws Exception {
+		assertEquals("foo.group.0", this.locator.locate("output.queue:foo"));
+	}
+
+	@Test
+	public void overrideNaturalOutputChannelName() throws Exception {
+		this.module.setOutputChannelName("bar");
+		assertEquals("foo.bar", this.locator.locate("output.queue:foo"));
+	}
+
+	@Test
+	public void noQueueQualifier() throws Exception {
+		assertEquals("foo.group.0", this.locator.locate("output.foo"));
+	}
+
+	@Test
+	public void underscoreSeparatorForChannelName() throws Exception {
+		assertEquals("foo.group.0", this.locator.locate("output_foo"));
+	}
+
+	@Test
+	public void overrideNaturalOutputChannelNamedQueue() throws Exception {
+		this.module.setOutputChannelName("queue:bar");
+		assertEquals("foo.bar", this.locator.locate("output.foo"));
+	}
+
+	@Test
+	public void overrideNaturalOutputChannelNamedQueueWithTopic() throws Exception {
+		this.module.setOutputChannelName("queue:bar");
+		assertEquals("topic:foo.bar", this.locator.locate("output.topic:foo"));
+	}
+
+	@Test
+	public void overrideNaturalOutputChannelNamedTopic() throws Exception {
+		this.module.setOutputChannelName("topic:bar");
+		assertEquals("foo.bar", this.locator.locate("output.queue:foo"));
+	}
+
+}
--- a/spring-cloud-streams/src/test/java/org/springframework/bus/runner/adapter/DiscoveryClientChannelLocatorTests.java
+++ b/spring-cloud-streams/src/test/java/org/springframework/bus/runner/adapter/DiscoveryClientChannelLocatorTests.java
@@ -0,0 +1,81 @@
+/*
+ * 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.bus.runner.adapter;
+
+import static org.junit.Assert.assertEquals;
+import static org.mockito.Matchers.any;
+
+import java.net.URI;
+import java.util.Arrays;
+import java.util.HashSet;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mockito;
+import org.springframework.bus.runner.adapter.discovery.DiscoveryClientChannelLocator;
+import org.springframework.bus.runner.config.MessageBusProperties;
+import org.springframework.cloud.client.DefaultServiceInstance;
+import org.springframework.cloud.client.discovery.DiscoveryClient;
+import org.springframework.web.client.RestOperations;
+
+/**
+ * @author Dave Syer
+ */
+public class DiscoveryClientChannelLocatorTests {
+
+	private DiscoveryClient client = Mockito.mock(DiscoveryClient.class);
+
+	private RestOperations restTemplate = Mockito.mock(RestOperations.class);
+
+	private DiscoveryClientChannelLocator locator = new DiscoveryClientChannelLocator(this.client, "service");
+
+	private ChannelsMetadata metadata = new ChannelsMetadata();
+
+	@Before
+	public void init() {
+		this.locator.setRestTemplate(this.restTemplate);
+		this.metadata.setModule(new MessageBusProperties());
+		this.metadata.setInputChannels(new HashSet<InputChannelBinding>());
+		this.metadata.setOutputChannels(new HashSet<OutputChannelBinding>());
+		Mockito.when(
+				this.restTemplate.getForObject(Mockito.any(URI.class), anyChannels()))
+				.thenReturn(this.metadata);
+		Mockito.when(this.client.getInstances(Mockito.anyString())).thenReturn(
+				Arrays.asList(new DefaultServiceInstance("service", "example.com", 888, false)));
+	}
+
+	@Test
+	public void locateInputFromOutput() {
+		OutputChannelBinding output = new OutputChannelBinding("output");
+		output.setRemoteName("foo.0");
+		this.metadata.getOutputChannels().add(output);
+		assertEquals("foo.0", this.locator.locate("input"));
+	}
+
+	@Test
+	public void locateOutputFromInput() {
+		InputChannelBinding input = new InputChannelBinding("input");
+		input.setRemoteName("foo.0");
+		this.metadata.getInputChannels().add(input);
+		assertEquals("foo.0", this.locator.locate("output"));
+	}
+
+	@SuppressWarnings({ "unchecked" })
+	private Class<ChannelsMetadata> anyChannels() {
+		return any(Class.class);
+	}
+}
--- a/spring-cloud-streams/src/test/java/org/springframework/bus/runner/config/MessageBusAdapterConfigurationTests.java
+++ b/spring-cloud-streams/src/test/java/org/springframework/bus/runner/config/MessageBusAdapterConfigurationTests.java
@@ -0,0 +1,150 @@
+/*
+ * 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.bus.runner.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.test.SpringApplicationConfiguration;
+import org.springframework.bus.runner.adapter.ChannelBinding;
+import org.springframework.bus.runner.adapter.MessageBusAdapter;
+import org.springframework.bus.runner.adapter.OutputChannelBinding;
+import org.springframework.bus.runner.config.MessageBusAdapterConfigurationTests.Empty;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
+import org.springframework.integration.channel.DirectChannel;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.annotation.DirtiesContext.ClassMode;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+import org.springframework.xd.dirt.integration.bus.local.LocalMessageBus;
+
+/**
+ * @author Dave Syer
+ */
+@RunWith(SpringJUnit4ClassRunner.class)
+@SpringApplicationConfiguration(classes = Empty.class)
+@DirtiesContext(classMode = ClassMode.AFTER_EACH_TEST_METHOD)
+public class MessageBusAdapterConfigurationTests {
+
+	@Autowired
+	private DefaultListableBeanFactory context;
+
+	@Autowired
+	private MessageBusAdapter adapter;
+
+	@Autowired
+	private MessageBusAdapterConfiguration configuration;
+
+	@Autowired
+	private MessageBusProperties module;
+
+	@Before
+	public void init() {
+	}
+
+	@Test
+	public void oneOutput() throws Exception {
+		this.context.registerSingleton("output", new DirectChannel());
+		refresh();
+		Collection<OutputChannelBinding> channels = this.adapter.getChannelsMetadata().getOutputChannels();
+		assertEquals(1, channels.size());
+		assertEquals("group.0", channels.iterator().next().getRemoteName());
+		assertEquals("tap:stream:group.module.0", channels.iterator().next().getTapChannelName());
+	}
+
+	private void refresh() {
+		Collection<OutputChannelBinding> channels = this.configuration.getOutputChannels();
+		for (OutputChannelBinding channel : channels) {
+			channel.setTapped(true);
+		}
+		this.adapter.setOutputChannels(channels);
+		this.adapter.start();
+	}
+
+	@Test
+	public void oneOutputTopic() throws Exception {
+		this.context.registerSingleton("output.topic:", new DirectChannel());
+		refresh();
+		Collection<OutputChannelBinding> channels = this.adapter.getChannelsMetadata().getOutputChannels();
+		assertEquals(1, channels.size());
+		assertEquals("topic:group.0", channels.iterator().next().getRemoteName());
+		assertEquals("tap:stream:group.module.0", channels.iterator().next().getTapChannelName());
+	}
+
+	@Test
+	public void twoOutputsWithQueue() throws Exception {
+		this.context.registerSingleton("output", new DirectChannel());
+		this.context.registerSingleton("output.queue:foo", new DirectChannel());
+		refresh();
+		Collection<OutputChannelBinding> channels = this.adapter.getChannelsMetadata().getOutputChannels();
+		List<String> names = getChannelNames(channels);
+		assertEquals(2, channels.size());
+		assertTrue(names.contains("group.0"));
+		assertTrue(names.contains("foo.group.0"));
+	}
+
+	private List<String> getChannelNames(Collection<? extends ChannelBinding> channels) {
+		List<String> list = new ArrayList<String>();
+		for (ChannelBinding binding : channels) {
+			list.add(binding.getRemoteName());
+		}
+		return list;
+	}
+
+	@Test
+	public void overrideNaturalOutputChannelName() throws Exception {
+		this.module.setOutputChannelName("bar");
+		this.context.registerSingleton("output.queue:foo", new DirectChannel());
+		refresh();
+		Collection<OutputChannelBinding> channels = this.adapter.getChannelsMetadata().getOutputChannels();
+		assertEquals(1, channels.size());
+		assertEquals("foo.bar", channels.iterator().next().getRemoteName());
+		// TODO: fix this. What should it be?
+		assertEquals("tap:stream:foo.bar.module.0", channels.iterator().next().getTapChannelName());
+	}
+
+	@Test
+	public void overrideNaturalOutputChannelNamedQueueWithTopic() throws Exception {
+		this.module.setOutputChannelName("queue:bar");
+		this.context.registerSingleton("output.topic:foo", new DirectChannel());
+		refresh();
+		Collection<OutputChannelBinding> channels = this.adapter.getChannelsMetadata().getOutputChannels();
+		assertEquals(1, channels.size());
+		assertEquals("topic:foo.bar", channels.iterator().next().getRemoteName());
+		assertEquals("tap:stream:foo.bar.module.0", channels.iterator().next().getTapChannelName());
+	}
+
+	@Configuration
+	@Import(MessageBusAdapterConfiguration.class)
+	protected static class Empty {
+		@Bean
+		public LocalMessageBus messageBus() {
+			return new LocalMessageBus();
+		}
+	}
+
+}