msg = MessageBuilder.withPayload(new TestPayload())
+ .setHeader(MessageHeaders.CONTENT_TYPE, MediaType.ALL_VALUE).build();
+
+ input.send(msg);
+ assertTrue(msgSent.get());
+ }
+
+ static class TestPayload {
+
+ @Override
+ public String toString() {
+ return "foo";
+ }
+
+ @Override
+ public boolean equals(Object other) {
+ return (other instanceof TestPayload && this.toString().equals(other.toString()));
+ }
+
+ @Override
+ public int hashCode() {
+ return this.toString().hashCode();
+ }
+
+ }
+}
diff --git a/spring-cloud-streams-bindings/spring-cloud-streams-binding-rabbit/pom.xml b/spring-cloud-streams-bindings/spring-cloud-streams-binding-rabbit/pom.xml
new file mode 100644
index 000000000..ea754bdfe
--- /dev/null
+++ b/spring-cloud-streams-bindings/spring-cloud-streams-binding-rabbit/pom.xml
@@ -0,0 +1,40 @@
+
+
+ 4.0.0
+
+ spring-cloud-streams-binding-rabbit
+ jar
+ spring-cloud-streams-binding-rabbit
+ RabbitMQ binding implementation
+
+
+ org.springframework.cloud
+ spring-cloud-streams-bindings-parent
+ 1.0.0.BUILD-SNAPSHOT
+
+
+
+ UTF-8
+
+
+
+
+ org.springframework.cloud
+ spring-cloud-streams-binding-spi
+
+
+ org.springframework.cloud
+ spring-cloud-streams-binding-test
+
+
+ org.springframework.boot
+ spring-boot-starter-amqp
+
+
+ org.springframework.integration
+ spring-integration-amqp
+ ${spring-integration.version}
+
+
+
diff --git a/spring-cloud-streams-bindings/spring-cloud-streams-binding-rabbit/src/main/java/org/springframework/xd/dirt/integration/rabbit/ConnectionFactorySettings.java b/spring-cloud-streams-bindings/spring-cloud-streams-binding-rabbit/src/main/java/org/springframework/xd/dirt/integration/rabbit/ConnectionFactorySettings.java
new file mode 100644
index 000000000..3822e7218
--- /dev/null
+++ b/spring-cloud-streams-bindings/spring-cloud-streams-binding-rabbit/src/main/java/org/springframework/xd/dirt/integration/rabbit/ConnectionFactorySettings.java
@@ -0,0 +1,84 @@
+/*
+ *
+ * * Copyright 2011-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.xd.dirt.integration.rabbit;
+
+import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.amqp.rabbit.connection.RabbitConnectionFactoryBean;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.autoconfigure.amqp.RabbitProperties;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.core.io.Resource;
+
+/**
+ * Configures the connection factory used by the rabbit message bus.
+ *
+ * @author Eric Bottard
+ * @author Gary Russell
+ */
+@Configuration
+public class ConnectionFactorySettings {
+
+ @Value("${spring.rabbitmq.useSSL:false}")
+ private boolean useSSL;
+
+ @Value("${spring.rabbitmq.sslProperties:}")
+ private Resource sslPropertiesLocation;
+
+ @Bean
+ // TODO: Move to spring boot
+ public ConnectionFactory rabbitConnectionFactory(RabbitProperties config,
+ com.rabbitmq.client.ConnectionFactory rabbitConnectionFactory) throws Exception {
+ CachingConnectionFactory factory = new CachingConnectionFactory(rabbitConnectionFactory);
+ factory.setAddresses(config.getAddresses());
+ if (config.getHost() != null) {
+ factory.setHost(config.getHost());
+ factory.setPort(config.getPort());
+ }
+ if (config.getUsername() != null) {
+ factory.setUsername(config.getUsername());
+ }
+ if (config.getPassword() != null) {
+ factory.setPassword(config.getPassword());
+ }
+ if (config.getVirtualHost() != null) {
+ factory.setVirtualHost(config.getVirtualHost());
+ }
+ return factory;
+ }
+
+ // If no RabbitProperties bean is available, instantiate one, deferring to Spring Boot for populating it
+ @Configuration
+ @ConditionalOnMissingBean(RabbitProperties.class)
+ @EnableConfigurationProperties(RabbitProperties.class)
+ public static class RabbitPropertiesLoader {
+ }
+
+ @Bean
+ public RabbitConnectionFactoryBean rabbitFactory() {
+ RabbitConnectionFactoryBean rabbitConnectionFactoryBean = new RabbitConnectionFactoryBean();
+ rabbitConnectionFactoryBean.setUseSSL(this.useSSL);
+ rabbitConnectionFactoryBean.setSslPropertiesLocation(this.sslPropertiesLocation);
+ return rabbitConnectionFactoryBean;
+ }
+
+}
diff --git a/spring-cloud-streams-bindings/spring-cloud-streams-binding-rabbit/src/main/java/org/springframework/xd/dirt/integration/rabbit/LocalizedQueueConnectionFactory.java b/spring-cloud-streams-bindings/spring-cloud-streams-binding-rabbit/src/main/java/org/springframework/xd/dirt/integration/rabbit/LocalizedQueueConnectionFactory.java
new file mode 100644
index 000000000..fda103988
--- /dev/null
+++ b/spring-cloud-streams-bindings/spring-cloud-streams-binding-rabbit/src/main/java/org/springframework/xd/dirt/integration/rabbit/LocalizedQueueConnectionFactory.java
@@ -0,0 +1,232 @@
+/*
+ * 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.xd.dirt.integration.rabbit;
+
+import java.net.URI;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.springframework.amqp.AmqpException;
+import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
+import org.springframework.amqp.rabbit.connection.Connection;
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.amqp.rabbit.connection.ConnectionListener;
+import org.springframework.amqp.rabbit.connection.RabbitConnectionFactoryBean;
+import org.springframework.amqp.rabbit.connection.RoutingConnectionFactory;
+import org.springframework.core.io.Resource;
+import org.springframework.util.Assert;
+import org.springframework.web.client.RestTemplate;
+import org.springframework.web.util.UriComponentsBuilder;
+import org.springframework.xd.dirt.integration.bus.RabbitManagementUtils;
+
+
+/**
+ * A {@link RoutingConnectionFactory} that determines the node on which a queue is located and
+ * returns a factory that connects directly to that node.
+ * The RabbitMQ management plugin is called over REST to determine the node and the corresponding
+ * address for that node is injected into the connection factory.
+ * A single instance of each connection factory is retained in a cache.
+ * If the location cannot be determined, the default connection factory is returned. This connection
+ * factory is typically configured to connect to all the servers in a fail-over mode.
+ * {@link #getTargetConnectionFactory(Object)} is invoked by the
+ * {@code SimpleMessageListenerContainer}, when establishing a connection, with the lookup key having
+ * the format {@code '[queueName]'}.
+ *
All {@link ConnectionFactory} methods delegate to the default
+ *
+ * @author Gary Russell
+ * @since 1.2
+ */
+public class LocalizedQueueConnectionFactory implements ConnectionFactory, RoutingConnectionFactory {
+
+ private final Log logger = LogFactory.getLog(getClass());
+
+ private final Map nodeFactories = new HashMap<>();
+
+ private final ConnectionFactory defaultConnectionFactory;
+
+ private final String[] addresses;
+
+ private final String[] adminAdresses;
+
+ private final String[] nodes;
+
+ private final String vhost;
+
+ private final String username;
+
+ private final String password;
+
+ private final boolean useSSL;
+
+ private final Resource sslPropertiesLocation;
+
+ /**
+ *
+ * @param defaultConnectionFactory the fallback connection factory to use if the queue can't be located.
+ * @param addresses the rabbitmq server addresses (host:port, ...).
+ * @param adminAddresses the rabbitmq admin addresses (http://host:port, ...) must be the same length
+ * as addresses.
+ * @param nodes the rabbitmq nodes corresponding to addresses (rabbit@server1, ...).
+ * @param vhost the virtual host.
+ * @param username the user name.
+ * @param password the password.
+ */
+ public LocalizedQueueConnectionFactory(ConnectionFactory defaultConnectionFactory,
+ String[] addresses, String[] adminAddresses, String[] nodes, String vhost,
+ String username, String password, boolean useSSL, Resource sslPropertiesLocation) {
+ Assert.isTrue(addresses.length == adminAddresses.length
+ && addresses.length == nodes.length,
+ "'addresses', 'adminAddresses', and 'nodes' properties must have equal length");
+ this.defaultConnectionFactory = defaultConnectionFactory;
+ this.addresses = Arrays.copyOf(addresses, addresses.length);
+ this.adminAdresses = Arrays.copyOf(adminAddresses, adminAddresses.length);
+ this.nodes = Arrays.copyOf(nodes, nodes.length);
+ this.vhost = vhost;
+ this.username = username;
+ this.password = password;
+ this.useSSL = useSSL;
+ this.sslPropertiesLocation = sslPropertiesLocation;
+ }
+
+ @Override
+ public Connection createConnection() throws AmqpException {
+ return this.defaultConnectionFactory.createConnection();
+ }
+
+ @Override
+ public String getHost() {
+ return this.defaultConnectionFactory.getHost();
+ }
+
+ @Override
+ public int getPort() {
+ return this.defaultConnectionFactory.getPort();
+ }
+
+ @Override
+ public String getVirtualHost() {
+ return this.vhost;
+ }
+
+ @Override
+ public void addConnectionListener(ConnectionListener listener) {
+ this.defaultConnectionFactory.addConnectionListener(listener);
+ }
+
+ @Override
+ public boolean removeConnectionListener(ConnectionListener listener) {
+ return this.defaultConnectionFactory.removeConnectionListener(listener);
+ }
+
+ @Override
+ public void clearConnectionListeners() {
+ this.defaultConnectionFactory.clearConnectionListeners();
+ }
+
+ @Override
+ public ConnectionFactory getTargetConnectionFactory(Object key) {
+ String queue = ((String) key);
+ queue = queue.substring(1, queue.length() - 1);
+ ConnectionFactory connectionFactory = determineConnectionFactory(queue);
+ if (connectionFactory == null) {
+ return this.defaultConnectionFactory;
+ }
+ else {
+ return connectionFactory;
+ }
+ }
+
+ private ConnectionFactory determineConnectionFactory(String queue) {
+ for (int i = 0; i < this.adminAdresses.length; i++) {
+ String adminUri = this.adminAdresses[i];
+ RestTemplate template = createRestTemplate(adminUri);
+ URI uri = UriComponentsBuilder.fromUriString(adminUri + "/api")
+ .pathSegment("queues", "{vhost}", "{queue}")
+ .buildAndExpand(this.vhost, queue).encode().toUri();
+ try {
+ @SuppressWarnings("unchecked")
+ Map queueInfo = template.getForObject(uri, Map.class);
+ if (queueInfo != null) {
+ String node = (String) queueInfo.get("node");
+ if (node != null) {
+ for (int j = 0; j < this.nodes.length; j++) {
+ if (this.nodes[j].equals(node)) {
+ return nodeConnectionFactory(queue, j);
+ }
+ }
+ }
+ }
+ }
+ catch (Exception e) {
+ logger.error("Failed to determine queue location for: " + queue + " at: " +
+ uri.toString(), e);
+ }
+ }
+ logger.warn("Failed to determine queue location for: " + queue);
+ return null;
+ }
+
+ private synchronized ConnectionFactory nodeConnectionFactory(String queue, int index) throws Exception {
+ String address = this.addresses[index];
+ String node = this.nodes[index];
+ if (logger.isDebugEnabled()) {
+ logger.debug("Queue: " + queue + " is on node: " + node + " at: " + address);
+ }
+ ConnectionFactory cf = this.nodeFactories.get(node);
+ if (cf == null) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Creating new connection factory for: " + address);
+ }
+ cf = createConnectionFactory(address);
+ this.nodeFactories.put(node, cf);
+ }
+ return cf;
+ }
+
+ /**
+ * Create a RestTemplate for the supplied URI.
+ * @param adminUri the URI.
+ * @return the template.
+ */
+ protected RestTemplate createRestTemplate(String adminUri) {
+ return RabbitManagementUtils.buildRestTemplate(adminUri, this.username, this.password);
+ }
+
+ /**
+ * Create a dedicated connection factory for the address.
+ * @param address the address to which the factory should connect.
+ * @return the connection factory.
+ * @throws Exception if errors occur during creation.
+ */
+ protected ConnectionFactory createConnectionFactory(String address) throws Exception {
+ RabbitConnectionFactoryBean rcfb = new RabbitConnectionFactoryBean();
+ rcfb.setUseSSL(this.useSSL);
+ rcfb.setSslPropertiesLocation(this.sslPropertiesLocation);
+ rcfb.afterPropertiesSet();
+ CachingConnectionFactory ccf = new CachingConnectionFactory(rcfb.getObject());
+ ccf.setAddresses(address);
+ ccf.setUsername(this.username);
+ ccf.setPassword(this.password);
+ ccf.setVirtualHost(this.vhost);
+ return ccf;
+ }
+
+}
diff --git a/spring-cloud-streams-bindings/spring-cloud-streams-binding-rabbit/src/main/java/org/springframework/xd/dirt/integration/rabbit/RabbitBusCleaner.java b/spring-cloud-streams-bindings/spring-cloud-streams-binding-rabbit/src/main/java/org/springframework/xd/dirt/integration/rabbit/RabbitBusCleaner.java
new file mode 100644
index 000000000..c739005cf
--- /dev/null
+++ b/spring-cloud-streams-bindings/spring-cloud-streams-binding-rabbit/src/main/java/org/springframework/xd/dirt/integration/rabbit/RabbitBusCleaner.java
@@ -0,0 +1,256 @@
+/*
+ * 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.xd.dirt.integration.rabbit;
+
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.springframework.web.client.RestTemplate;
+import org.springframework.web.util.UriComponentsBuilder;
+import org.springframework.xd.dirt.integration.bus.BusCleaner;
+import org.springframework.xd.dirt.integration.bus.BusUtils;
+import org.springframework.xd.dirt.integration.bus.MessageBusSupport;
+import org.springframework.xd.dirt.integration.bus.RabbitAdminException;
+import org.springframework.xd.dirt.integration.bus.RabbitManagementUtils;
+
+
+/**
+ * Implementation of {@link org.springframework.xd.dirt.integration.bus.BusCleaner} for the {@code RabbitMessageBus}.
+ * @author Gary Russell
+ * @author David Turanski
+ * @since 1.2
+ */
+public class RabbitBusCleaner implements BusCleaner {
+
+ private final static Logger logger = LoggerFactory.getLogger(RabbitBusCleaner.class);
+
+ @Override
+ public Map> clean(String entity, boolean isJob) {
+ return clean("http://localhost:15672", "guest", "guest", "/", "xdbus.", entity, isJob);
+ }
+
+ public Map> clean(String adminUri, String user, String pw, String vhost,
+ String busPrefix, String entity, boolean isJob) {
+ return doClean(
+ adminUri == null ? "http://localhost:15672" : adminUri,
+ user == null ? "guest" : user,
+ pw == null ? "guest" : pw,
+ vhost == null ? "/" : vhost,
+ busPrefix == null ? "xdbus." : busPrefix,
+ entity, isJob);
+ }
+
+ private Map> doClean(String adminUri, String user, String pw, String vhost,
+ String busPrefix, String entity, boolean isJob) {
+ RestTemplate restTemplate = RabbitManagementUtils.buildRestTemplate(adminUri, user, pw);
+ List removedQueues = isJob
+ ? null//findJobQueues(adminUri, vhost, busPrefix, entity, restTemplate)
+ : findStreamQueues(adminUri, vhost, busPrefix, entity, restTemplate);
+ ExchangeCandidateCallback callback = null;
+ if (isJob) {
+// String pattern;
+// if (entity.endsWith("*")) {
+// pattern = entity.substring(0, entity.length() - 1) + "[^.]*";
+// }
+// else {
+// pattern = entity;
+// }
+// Collection exchangeNames = JobEventsListenerPlugin.getEventListenerChannels(pattern).values();
+// final Set jobExchanges = new HashSet<>();
+// for (String exchange : exchangeNames) {
+// jobExchanges.add(Pattern.compile(MessageBusSupport.applyPrefix(busPrefix,
+// MessageBusSupport.applyPubSub(exchange))));
+// }
+// jobExchanges.add(Pattern.compile(MessageBusSupport.applyPrefix(busPrefix, MessageBusSupport.applyPubSub(
+// JobEventsListenerPlugin.getEventListenerChannelName(pattern)))));
+// callback = new ExchangeCandidateCallback() {
+//
+// @Override
+// public boolean isCandidate(String exchangeName) {
+// for (Pattern pattern : jobExchanges) {
+// Matcher matcher = pattern.matcher(exchangeName);
+// if (matcher.matches()) {
+// return true;
+// }
+// }
+// return false;
+// }
+//
+// };
+ }
+ else {
+ final String tapPrefix = adjustPrefix(MessageBusSupport.applyPrefix(busPrefix,
+ MessageBusSupport.applyPubSub(BusUtils.constructTapPrefix(entity))));
+ callback = new ExchangeCandidateCallback() {
+
+ @Override
+ public boolean isCandidate(String exchangeName) {
+ return exchangeName.startsWith(tapPrefix);
+ }
+ };
+ }
+ List removedExchanges = findExchanges(adminUri, vhost, busPrefix, entity, restTemplate, callback);
+ // Delete the queues in reverse order to enable re-running after a partial success.
+ // The queue search above starts with 0 and terminates on a not found.
+ for (int i = removedQueues.size() - 1; i >= 0; i--) {
+ String queueName = removedQueues.get(i);
+ URI uri = UriComponentsBuilder.fromUriString(adminUri + "/api")
+ .pathSegment("queues", "{vhost}", "{stream}")
+ .buildAndExpand(vhost, queueName).encode().toUri();
+ restTemplate.delete(uri);
+ if (logger.isDebugEnabled()) {
+ logger.debug("deleted queue: " + queueName);
+ }
+ }
+ Map> results = new HashMap<>();
+ if (removedQueues.size() > 0) {
+ results.put("queues", removedQueues);
+ }
+ // Fanout exchanges for taps
+ for (String exchange : removedExchanges) {
+ URI uri = UriComponentsBuilder.fromUriString(adminUri + "/api")
+ .pathSegment("exchanges", "{vhost}", "{name}")
+ .buildAndExpand(vhost, exchange).encode().toUri();
+ restTemplate.delete(uri);
+ if (logger.isDebugEnabled()) {
+ logger.debug("deleted exchange: " + exchange);
+ }
+ }
+ if (removedExchanges.size() > 0) {
+ results.put("exchanges", removedExchanges);
+ }
+ return results;
+ }
+
+ private List findStreamQueues(String adminUri, String vhost, String busPrefix, String stream,
+ RestTemplate restTemplate) {
+ String queueNamePrefix = adjustPrefix(MessageBusSupport.applyPrefix(busPrefix, stream));
+ List> queues = listAllQueues(adminUri, vhost, restTemplate);
+ List removedQueues = new ArrayList<>();
+ for (Map queue : queues) {
+ String queueName = (String) queue.get("name");
+ if (queueName.startsWith(queueNamePrefix)) {
+ checkNoConsumers(queueName, queue);
+ removedQueues.add(queueName);
+ }
+ }
+ return removedQueues;
+ }
+
+// private List findJobQueues(String adminUri, String vhost, String busPrefix, String job,
+// RestTemplate restTemplate) {
+// List removedQueues = new ArrayList<>();
+// String jobQueueName = MessageBusSupport.applyPrefix(busPrefix,
+// AbstractJobPlugin.getJobChannelName(job));
+// String jobRequestsQueuePrefix = adjustPrefix(MessageBusSupport.applyPrefix(busPrefix,
+// AbstractJobPlugin.getJobChannelName(job)));
+// List> queues = listAllQueues(adminUri, vhost, restTemplate);
+// for (Map queue : queues) {
+// String queueName = (String) queue.get("name");
+// if (job.endsWith("*")) {
+// if (queueName.startsWith(jobQueueName.substring(0, jobQueueName.length() - 1))) {
+// checkNoConsumers(queueName, queue);
+// removedQueues.add(queueName);
+// }
+// }
+// else {
+// if (queueName.equals(jobQueueName)) {
+// checkNoConsumers(queueName, queue);
+// removedQueues.add(queueName);
+// }
+// else if (queueName.startsWith(jobRequestsQueuePrefix)
+// && queueName.endsWith(MessageBusSupport.applyRequests(""))) {
+// checkNoConsumers(queueName, queue);
+// removedQueues.add(queueName);
+// }
+// }
+// }
+// return removedQueues;
+// }
+
+ private List> listAllQueues(String adminUri, String vhost, RestTemplate restTemplate) {
+ URI uri = UriComponentsBuilder.fromUriString(adminUri + "/api")
+ .pathSegment("queues", "{vhost}")
+ .buildAndExpand(vhost).encode().toUri();
+ @SuppressWarnings("unchecked")
+ List> queues = restTemplate.getForObject(uri, List.class);
+ return queues;
+ }
+
+ private String adjustPrefix(String prefix) {
+ if (prefix.endsWith("*")) {
+ return prefix.substring(0, prefix.length() - 1);
+ }
+ else {
+ return prefix + BusUtils.GROUP_INDEX_DELIMITER;
+ }
+ }
+
+ private void checkNoConsumers(String queueName, Map queue) {
+ if (!queue.get("consumers").equals(Integer.valueOf(0))) {
+ throw new RabbitAdminException("Queue " + queueName + " is in use");
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private List findExchanges(String adminUri, String vhost, String busPrefix, String entity,
+ RestTemplate restTemplate, ExchangeCandidateCallback callback) {
+ List removedExchanges = new ArrayList<>();
+ URI uri = UriComponentsBuilder.fromUriString(adminUri + "/api")
+ .pathSegment("exchanges", "{vhost}")
+ .buildAndExpand(vhost).encode().toUri();
+ List> exchanges = restTemplate.getForObject(uri, List.class);
+ for (Map exchange : exchanges) {
+ String exchangeName = (String) exchange.get("name");
+ if (callback.isCandidate(exchangeName)) {
+ uri = UriComponentsBuilder.fromUriString(adminUri + "/api")
+ .pathSegment("exchanges", "{vhost}", "{name}", "bindings", "source")
+ .buildAndExpand(vhost, exchangeName).encode().toUri();
+ List> bindings = restTemplate.getForObject(uri, List.class);
+ if (bindings.size() == 0) {
+ uri = UriComponentsBuilder.fromUriString(adminUri + "/api")
+ .pathSegment("exchanges", "{vhost}", "{name}", "bindings", "destination")
+ .buildAndExpand(vhost, exchangeName).encode().toUri();
+ bindings = restTemplate.getForObject(uri, List.class);
+ if (bindings.size() == 0) {
+ removedExchanges.add((String) exchange.get("name"));
+ }
+ else {
+ throw new RabbitAdminException("Cannot delete exchange " + exchangeName
+ + "; it is a destination: " + bindings);
+ }
+ }
+ else {
+ throw new RabbitAdminException("Cannot delete exchange " + exchangeName + "; it has bindings: "
+ + bindings);
+ }
+ }
+ }
+ return removedExchanges;
+ }
+
+ private interface ExchangeCandidateCallback {
+
+ boolean isCandidate(String exchangeName);
+ }
+
+}
diff --git a/spring-cloud-streams-bindings/spring-cloud-streams-binding-rabbit/src/main/java/org/springframework/xd/dirt/integration/rabbit/RabbitMessageBus.java b/spring-cloud-streams-bindings/spring-cloud-streams-binding-rabbit/src/main/java/org/springframework/xd/dirt/integration/rabbit/RabbitMessageBus.java
new file mode 100644
index 000000000..073b43e2e
--- /dev/null
+++ b/spring-cloud-streams-bindings/spring-cloud-streams-binding-rabbit/src/main/java/org/springframework/xd/dirt/integration/rabbit/RabbitMessageBus.java
@@ -0,0 +1,1066 @@
+/*
+ * Copyright 2013-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.xd.dirt.integration.rabbit;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+
+import org.aopalliance.aop.Advice;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.springframework.amqp.core.AcknowledgeMode;
+import org.springframework.amqp.core.BindingBuilder;
+import org.springframework.amqp.core.DirectExchange;
+import org.springframework.amqp.core.Exchange;
+import org.springframework.amqp.core.FanoutExchange;
+import org.springframework.amqp.core.MessageDeliveryMode;
+import org.springframework.amqp.core.MessagePostProcessor;
+import org.springframework.amqp.core.MessageProperties;
+import org.springframework.amqp.core.Queue;
+import org.springframework.amqp.rabbit.config.RetryInterceptorBuilder;
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.amqp.rabbit.core.BatchingRabbitTemplate;
+import org.springframework.amqp.rabbit.core.ChannelCallback;
+import org.springframework.amqp.rabbit.core.RabbitAdmin;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.amqp.rabbit.core.support.BatchingStrategy;
+import org.springframework.amqp.rabbit.core.support.SimpleBatchingStrategy;
+import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
+import org.springframework.amqp.rabbit.retry.MessageRecoverer;
+import org.springframework.amqp.rabbit.retry.RejectAndDontRequeueRecoverer;
+import org.springframework.amqp.rabbit.retry.RepublishMessageRecoverer;
+import org.springframework.amqp.rabbit.support.DefaultMessagePropertiesConverter;
+import org.springframework.amqp.rabbit.support.MessagePropertiesConverter;
+import org.springframework.amqp.support.AmqpHeaders;
+import org.springframework.amqp.support.postprocessor.DelegatingDecompressingPostProcessor;
+import org.springframework.amqp.support.postprocessor.GZipPostProcessor;
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
+import org.springframework.beans.factory.support.DefaultListableBeanFactory;
+import org.springframework.context.Lifecycle;
+import org.springframework.context.support.GenericApplicationContext;
+import org.springframework.core.io.Resource;
+import org.springframework.core.task.SimpleAsyncTaskExecutor;
+import org.springframework.expression.Expression;
+import org.springframework.expression.ExpressionParser;
+import org.springframework.expression.spel.standard.SpelExpressionParser;
+import org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter;
+import org.springframework.integration.amqp.outbound.AmqpOutboundEndpoint;
+import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper;
+import org.springframework.integration.channel.DirectChannel;
+import org.springframework.integration.context.IntegrationContextUtils;
+import org.springframework.integration.endpoint.EventDrivenConsumer;
+import org.springframework.integration.handler.AbstractMessageHandler;
+import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.MessageHandler;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.messaging.SubscribableChannel;
+import org.springframework.retry.interceptor.RetryOperationsInterceptor;
+import org.springframework.scheduling.TaskScheduler;
+import org.springframework.util.Assert;
+import org.springframework.util.ClassUtils;
+import org.springframework.util.StringUtils;
+import org.springframework.xd.dirt.integration.bus.AbstractBusPropertiesAccessor;
+import org.springframework.xd.dirt.integration.bus.Binding;
+import org.springframework.xd.dirt.integration.bus.BusProperties;
+import org.springframework.xd.dirt.integration.bus.BusUtils;
+import org.springframework.xd.dirt.integration.bus.MessageBus;
+import org.springframework.xd.dirt.integration.bus.MessageBusSupport;
+import org.springframework.xd.dirt.integration.bus.MessageValues;
+import org.springframework.xd.dirt.integration.bus.serializer.MultiTypeCodec;
+
+import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.Envelope;
+
+/**
+ * A {@link MessageBus} implementation backed by RabbitMQ.
+ * @author Mark Fisher
+ * @author Gary Russell
+ * @author Jennifer Hickey
+ * @author Gunnar Hillert
+ * @author Ilayaperumal Gopinathan
+ * @author David Turanski
+ */
+public class RabbitMessageBus extends MessageBusSupport implements DisposableBean {
+
+ private static final AcknowledgeMode DEFAULT_ACKNOWLEDGE_MODE = AcknowledgeMode.AUTO;
+
+ private static final MessageDeliveryMode DEFAULT_DEFAULT_DELIVERY_MODE = MessageDeliveryMode.PERSISTENT;
+
+ private static final boolean DEFAULT_DEFAULT_REQUEUE_REJECTED = true;
+
+ private static final int DEFAULT_MAX_CONCURRENCY = 1;
+
+ private static final int DEFAULT_PREFETCH_COUNT = 1;
+
+ private static final String DEFAULT_RABBIT_PREFIX = "xdbus.";
+
+ private static final int DEFAULT_TX_SIZE = 1;
+
+ private static final String[] DEFAULT_REQUEST_HEADER_PATTERNS = new String[] { "STANDARD_REQUEST_HEADERS", "*" };
+
+ private static final String[] DEFAULT_REPLY_HEADER_PATTERNS = new String[] { "STANDARD_REPLY_HEADERS", "*" };
+
+ private static final String DEAD_LETTER_EXCHANGE = "DLX";
+
+ private static final Set