GH-203: Support Late Binding with RabbitMQ

If the broker was down, the binding failed. Defer the queue/exchange/binding declarations
until the broker is available (already supported by `RabbitAdmin`.

Solves spring-cloud/spring-cloud-stream#203

Fix PubSub, Test

- Too many binder prefixes on pub sub queue name.
- Mismatched queue args in test case.

spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java

@@ -30,6 +30,7 @@ import org.aopalliance.aop.Advice;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.springframework.amqp.AmqpConnectException;
 import org.springframework.amqp.core.AcknowledgeMode;
 import org.springframework.amqp.core.BindingBuilder;
 import org.springframework.amqp.core.DirectExchange;
@@ -42,7 +43,6 @@ import org.springframework.amqp.core.TopicExchange;
 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;
@@ -309,6 +309,7 @@ public class RabbitMessageChannelBinder extends MessageChannelBinderSupport impl
 		this.rabbitAdmin = new RabbitAdmin(connectionFactory);
 		this.autoDeclareContext.refresh();
 		this.rabbitAdmin.setApplicationContext(this.autoDeclareContext);
+		this.rabbitAdmin.setIgnoreDeclarationExceptions(true);
 		this.rabbitAdmin.afterPropertiesSet();
 	}
 
@@ -447,17 +448,19 @@ public class RabbitMessageChannelBinder extends MessageChannelBinderSupport impl
 		RabbitPropertiesAccessor accessor = new RabbitPropertiesAccessor(properties);
 		String queueName = applyPrefix(accessor.getPrefix(this.defaultPrefix), name);
 		TopicExchange exchange = new TopicExchange(queueName);
-		declareExchangeIfNotPresent(exchange);
+		declareExchange(queueName, exchange);
 		int partitionIndex = accessor.getPartitionIndex();
+		String dlqNameRoot = name;
 		if (partitionIndex >= 0) {
-			queueName += "-" + partitionIndex;
+			String partitionSuffix = "-" + partitionIndex;
+			queueName += partitionSuffix;
+			dlqNameRoot += partitionSuffix;
 		}
-		Map<String, Object> args = queueArgs(accessor, queueName);
-		Queue queue = new Queue(queueName, true, false, false, args);
-		declareQueueIfNotPresent(queue);
-		autoBindDLQ(name, accessor);
+		Queue queue = new Queue(queueName, true, false, false, queueArgs(accessor, queueName));
+		declareQueue(queueName, queue);
+		autoBindDLQ(dlqNameRoot, accessor);
 		org.springframework.amqp.core.Binding binding = BindingBuilder.bind(queue).to(exchange).with(queueName);
-		this.rabbitAdmin.declareBinding(binding);
+		declareBinding(queueName, binding);
 		doRegisterConsumer(name, moduleInputChannel, queue, accessor, false);
 		bindExistingProducerDirectlyIfPossible(name, moduleInputChannel);
 	}
@@ -473,28 +476,18 @@ public class RabbitMessageChannelBinder extends MessageChannelBinderSupport impl
 		validateConsumerProperties(name, properties, SUPPORTED_PUBSUB_CONSUMER_PROPERTIES);
 		String prefix = accessor.getPrefix(this.defaultPrefix);
 		TopicExchange exchange = new TopicExchange(applyPrefix(prefix, exchangeName));
-		declareExchangeIfNotPresent(exchange);
+		declareExchange(exchange.getName(), exchange);
 		Queue queue;
 		boolean durable = accessor.isDurable(this.defaultDurableSubscription);
 		String queueName = applyPrefix(prefix, name);
 		if (durable) {
-			Map<String, Object> args = queueArgs(accessor, queueName);
-			queue = new Queue(queueName, true, false, false, args);
+			queue = new Queue(queueName, true, false, false, queueArgs(accessor, queueName));
 		}
 		else {
 			queue = new Queue(queueName, false, false, true);
 		}
-		declareQueueIfNotPresent(queue);
-		org.springframework.amqp.core.Binding binding = BindingBuilder.bind(queue).to(exchange).with("#");
-		this.rabbitAdmin.declareBinding(binding);
-		// register with context so they will be redeclared after a connection failure if auto-delete
-		if (!durable && !this.autoDeclareContext.containsBean(name)) {
-			this.autoDeclareContext.getBeanFactory().registerSingleton(name, queue);
-		}
-		String bindingBeanName = exchange.getName() + "." + queue.getName() + ".binding";
-		if (!durable && !this.autoDeclareContext.containsBean(bindingBeanName)) {
-			this.autoDeclareContext.getBeanFactory().registerSingleton(bindingBeanName, binding);
-		}
+		declareQueue(queueName, queue);
+		declareBinding(queue.getName(), BindingBuilder.bind(queue).to(exchange).with("#"));
 		doRegisterConsumer(name, moduleInputChannel, queue, accessor, true);
 		if (durable) {
 			autoBindDLQ(name, accessor);
@@ -586,7 +579,6 @@ public class RabbitMessageChannelBinder extends MessageChannelBinderSupport impl
 			RepublishMessageRecoverer republishMessageRecoverer = new RepublishMessageRecoverer(errorTemplate,
 					deadLetterExchangeName(prefix),
 					applyPrefix(prefix, name));
-			// TODO: Add container id to republished message headers? (Needs AMQP-489).
 			return republishMessageRecoverer;
 		}
 		else {
@@ -621,26 +613,27 @@ public class RabbitMessageChannelBinder extends MessageChannelBinderSupport impl
 		String partitionKeyExtractorClass = properties.getPartitionKeyExtractorClass();
 		Expression partitionKeyExpression = properties.getPartitionKeyExpression();
 		TopicExchange exchange = new TopicExchange(queueName);
-		declareExchangeIfNotPresent(exchange);
+		declareExchange(queueName, exchange);
 		AmqpOutboundEndpoint endpoint = new AmqpOutboundEndpoint(rabbitTemplate);
 		endpoint.setExchangeName(exchange.getName());
 		if (partitionKeyExpression == null && !StringUtils.hasText(partitionKeyExtractorClass)) {
-			Queue queue = new Queue(queueName);
-			declareQueueIfNotPresent(queue);
+			Queue queue = new Queue(queueName, true, false, false, queueArgs(properties, queueName));
+			declareQueue(name, queue);
+			autoBindDLQ(name, properties);
 			endpoint.setRoutingKey(queueName);
 			org.springframework.amqp.core.Binding binding = BindingBuilder.bind(queue).to(exchange).with(queueName);
-			this.rabbitAdmin.declareBinding(binding);
+			declareBinding(queueName, binding);
 		}
 		else {
 			endpoint.setExpressionRoutingKey(EXPRESSION_PARSER.parseExpression(buildPartitionRoutingExpression
 					(queueName)));
 			// if the stream is partitioned, create one queue for each target partition
 			for (int i = 0; i < properties.getNextModuleCount(); i++) {
-				Queue queue = new Queue(queueName + "-" + i);
-				this.rabbitAdmin.declareQueue(queue);
-				org.springframework.amqp.core.Binding binding = BindingBuilder.bind(queue).to(exchange)
-						.with(queue.getName());
-				this.rabbitAdmin.declareBinding(binding);
+				String partitionSuffix = "-" + i;
+				Queue queue = new Queue(queueName + partitionSuffix, true, false, false, queueArgs(properties, queueName));
+				declareQueue(queue.getName(), queue);
+				autoBindDLQ(name + partitionSuffix, properties);
+				declareBinding(queue.getName(), BindingBuilder.bind(queue).to(exchange).with(queue.getName()));
 			}
 		}
 		configureOutboundHandler(endpoint, properties);
@@ -663,7 +656,8 @@ public class RabbitMessageChannelBinder extends MessageChannelBinderSupport impl
 		validateProducerProperties(name, properties, SUPPORTED_PUBSUB_PRODUCER_PROPERTIES);
 		RabbitPropertiesAccessor accessor = new RabbitPropertiesAccessor(properties);
 		String exchangeName = applyPrefix(accessor.getPrefix(this.defaultPrefix), name);
-		declareExchangeIfNotPresent(new TopicExchange(exchangeName));
+		TopicExchange exchange = new TopicExchange(exchangeName);
+		declareExchange(exchangeName, exchange);
 		AmqpOutboundEndpoint endpoint = new AmqpOutboundEndpoint(determineRabbitTemplate(accessor));
 		endpoint.setExchangeName(exchangeName);
 		endpoint.setRoutingKey(name);
@@ -731,11 +725,7 @@ public class RabbitMessageChannelBinder extends MessageChannelBinderSupport impl
 				+ this.getIdGenerator().generateId();
 		this.doRegisterProducer(name, requests, queue, replyQueueName, accessor);
 		Queue replyQueue = new Queue(replyQueueName, false, false, true); // auto-delete
-		declareQueueIfNotPresent(replyQueue);
-		// register with context so it will be redeclared after a connection failure
-		if (!this.autoDeclareContext.containsBean(replyQueueName)) {
-			this.autoDeclareContext.getBeanFactory().registerSingleton(replyQueueName, replyQueue);
-		}
+		declareQueue(replyQueueName, replyQueue);
 		this.doRegisterConsumer(name, replies, replyQueue, accessor, false);
 	}
 
@@ -748,7 +738,7 @@ public class RabbitMessageChannelBinder extends MessageChannelBinderSupport impl
 		validateConsumerProperties(name, properties, SUPPORTED_REPLYING_CONSUMER_PROPERTIES);
 		RabbitPropertiesAccessor accessor = new RabbitPropertiesAccessor(properties);
 		Queue requestQueue = new Queue(applyPrefix(accessor.getPrefix(this.defaultPrefix), applyRequests(name)));
-		declareQueueIfNotPresent(requestQueue);
+		declareQueue(requestQueue.getName(), requestQueue);
 		this.doRegisterConsumer(name, requests, requestQueue, accessor, false);
 
 		AmqpOutboundEndpoint replyQueue = new AmqpOutboundEndpoint(this.rabbitTemplate);
@@ -758,37 +748,6 @@ public class RabbitMessageChannelBinder extends MessageChannelBinderSupport impl
 		doRegisterProducer(name, replies, replyQueue, accessor);
 	}
 
-	/**
-	 * Try passive declaration first, in case the user has pre-configured the queue with incompatible arguments.
-	 * @param queue The queue.
-	 */
-	private void declareQueueIfNotPresent(Queue queue) {
-		if (this.rabbitAdmin.getQueueProperties(queue.getName()) == null) {
-			this.rabbitAdmin.declareQueue(queue);
-		}
-	}
-
-	/**
-	 * Try passive declaration first, in case the user has pre-configured the exchange with incompatible arguments.
-	 * @param exchange
-	 */
-	private void declareExchangeIfNotPresent(final Exchange exchange) {
-		this.rabbitTemplate.execute(new ChannelCallback<Void>() {
-
-			@Override
-			public Void doInRabbit(Channel channel) throws Exception {
-				try {
-					channel.exchangeDeclarePassive(exchange.getName());
-				}
-				catch (IOException e) {
-					RabbitMessageChannelBinder.this.rabbitAdmin.declareExchange(exchange);
-				}
-				return null;
-			}
-
-		});
-	}
-
 	/**
 	 * If so requested, declare the DLX/DLQ and bind it. The DLQ is bound to the DLX with a routing key of the original
 	 * queue name because we use default exchange routing by queue name for the original message.
@@ -805,14 +764,50 @@ public class RabbitMessageChannelBinder extends MessageChannelBinderSupport impl
 			String queueName = applyPrefix(prefix, name);
 			String dlqName = constructDLQName(queueName);
 			Queue dlq = new Queue(dlqName);
-			declareQueueIfNotPresent(dlq);
+			declareQueue(dlqName, dlq);
 			final String dlxName = deadLetterExchangeName(prefix);
 			final DirectExchange dlx = new DirectExchange(dlxName);
-			declareExchangeIfNotPresent(dlx);
-			this.rabbitAdmin.declareBinding(BindingBuilder.bind(dlq).to(dlx).with(queueName));
+			declareExchange(dlxName, dlx);
+			declareBinding(dlqName, BindingBuilder.bind(dlq).to(dlx).with(queueName));
 		}
 	}
 
+	public void declareQueue(String beanName, Queue queue) {
+		try {
+			this.rabbitAdmin.declareQueue(queue);
+		}
+		catch (AmqpConnectException e) {
+			if (logger.isDebugEnabled()) {
+				logger.debug("Declaration of queue: " + queue.getName() + " deferred - connection not available");
+			}
+		}
+		addToAutoDeclareContext(beanName, queue);
+	}
+
+	public void declareExchange(final String rootName, final Exchange exchange) {
+		try {
+			this.rabbitAdmin.declareExchange(exchange);
+		}
+		catch (AmqpConnectException e) {
+			if (logger.isDebugEnabled()) {
+				logger.debug("Declaration of exchange: " + exchange.getName() + " deferred - connection not available");
+			}
+		}
+		addToAutoDeclareContext(rootName + ".exchange", exchange);
+	}
+
+	public void declareBinding(String rootName, org.springframework.amqp.core.Binding binding) {
+		try {
+			this.rabbitAdmin.declareBinding(binding);
+		}
+		catch (AmqpConnectException e) {
+			if (logger.isDebugEnabled()) {
+				logger.debug("Declaration of binding: " + rootName + ".binding deferred - connection not available");
+			}
+		}
+		addToAutoDeclareContext(rootName + ".binding", binding);
+	}
+
 	private String deadLetterExchangeName(String prefix) {
 		return prefix + DEAD_LETTER_EXCHANGE;
 	}
@@ -835,11 +830,21 @@ public class RabbitMessageChannelBinder extends MessageChannelBinderSupport impl
 		cleanAutoDeclareContext(BinderUtils.groupedName(name, group));
 	}
 
+	private void addToAutoDeclareContext(String name, Object bean) {
+		synchronized (this.autoDeclareContext) {
+			if (!this.autoDeclareContext.containsBean(name)) {
+				this.autoDeclareContext.getBeanFactory().registerSingleton(name, bean);
+			}
+		}
+	}
+
 	private void cleanAutoDeclareContext(String name) {
-		if (this.autoDeclareContext.containsBean(name)) {
-			ConfigurableListableBeanFactory beanFactory = this.autoDeclareContext.getBeanFactory();
-			if (beanFactory instanceof DefaultListableBeanFactory) {
-				((DefaultListableBeanFactory) beanFactory).destroySingleton(name);
+		synchronized(this.autoDeclareContext) {
+			if (this.autoDeclareContext.containsBean(name)) {
+				ConfigurableListableBeanFactory beanFactory = this.autoDeclareContext.getBeanFactory();
+				if (beanFactory instanceof DefaultListableBeanFactory) {
+					((DefaultListableBeanFactory) beanFactory).destroySingleton(name);
+				}
 			}
 		}
 	}

spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java

@@ -27,6 +27,7 @@ import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
+import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
@@ -49,6 +50,7 @@ import org.mockito.ArgumentCaptor;
 import org.springframework.amqp.core.AcknowledgeMode;
 import org.springframework.amqp.core.MessageDeliveryMode;
 import org.springframework.amqp.core.Queue;
+import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
 import org.springframework.amqp.rabbit.core.RabbitAdmin;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
@@ -56,12 +58,14 @@ import org.springframework.amqp.support.AmqpHeaders;
 import org.springframework.amqp.support.postprocessor.DelegatingDecompressingPostProcessor;
 import org.springframework.amqp.utils.test.TestUtils;
 import org.springframework.beans.DirectFieldAccessor;
+import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
 import org.springframework.cloud.stream.binder.Binder;
 import org.springframework.cloud.stream.binder.BinderProperties;
 import org.springframework.cloud.stream.binder.Binding;
 import org.springframework.cloud.stream.binder.PartitionCapableBinderTests;
 import org.springframework.cloud.stream.binder.Spy;
 import org.springframework.cloud.stream.test.junit.rabbit.RabbitTestSupport;
+import org.springframework.context.support.AbstractApplicationContext;
 import org.springframework.expression.spel.standard.SpelExpression;
 import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.channel.QueueChannel;
@@ -529,6 +533,7 @@ public class RabbitBinderTests extends PartitionCapableBinderTests {
 		RabbitAdmin admin = new RabbitAdmin(this.rabbitAvailableRule.getResource());
 		Map<String, Object> args = new HashMap<String, Object>();
 		args.put("x-dead-letter-exchange", TEST_PREFIX + "DLX");
+		args.put("x-dead-letter-routing-key", TEST_PREFIX + "dlqpubtest");
 		Queue queue = new Queue(TEST_PREFIX + "dlqpubtest", true, false, false, args);
 		admin.declareQueue(queue);
 
@@ -629,6 +634,116 @@ public class RabbitBinderTests extends PartitionCapableBinderTests {
 		binder.unbindConsumers("batching.0");
 	}
 
+	/*
+	 * Test late binding due to broker down; queues with and without DLQs, and
+	 * partitioned queues.
+	 */
+	@Test
+	public void testLateBinding() throws Exception {
+		RabbitTestSupport.RabbitProxy proxy = new RabbitTestSupport.RabbitProxy();
+		CachingConnectionFactory cf = new CachingConnectionFactory("localhost", proxy.getPort());
+		RabbitMessageChannelBinder binder = new RabbitMessageChannelBinder(cf);
+		AbstractApplicationContext applicationContext = mock(AbstractApplicationContext.class);
+		when(applicationContext.getBeanFactory()).thenReturn(mock(ConfigurableListableBeanFactory.class));
+		binder.setApplicationContext(applicationContext);
+		binder.setDefaultAutoBindDLQ(true);
+		binder.afterPropertiesSet();
+		Properties properties = new Properties();
+		properties.put("prefix", "latebinder.");
+
+		MessageChannel moduleOutputChannel = new DirectChannel();
+		binder.bindProducer("late.0", moduleOutputChannel, properties);
+
+		QueueChannel moduleInputChannel = new QueueChannel();
+		binder.bindConsumer("late.0", moduleInputChannel, properties);
+
+		properties.put("partitionKeyExpression", "payload.equals('0') ? 0 : 1");
+		properties.put("partitionSelectorExpression", "hashCode()");
+		properties.put("nextModuleCount", "2");
+
+		MessageChannel partOutputChannel = new DirectChannel();
+		binder.bindProducer("partlate.0", partOutputChannel, properties);
+
+		QueueChannel partInputChannel0 = new QueueChannel();
+		QueueChannel partInputChannel1 = new QueueChannel();
+		properties.clear();
+		properties.put("prefix", "latebinder.");
+		properties.put("partitionIndex", "0");
+		binder.bindConsumer("partlate.0", partInputChannel0, properties);
+		properties.put("partitionIndex", "1");
+		binder.bindConsumer("partlate.0", partInputChannel1, properties);
+
+		binder.setDefaultAutoBindDLQ(false);
+		properties.clear();
+		properties.put("prefix", "latebinder.");
+		MessageChannel noDLQOutputChannel = new DirectChannel();
+		binder.bindProducer("lateNoDLQ.0", noDLQOutputChannel, properties);
+
+		QueueChannel noDLQInputChannel = new QueueChannel();
+		binder.bindConsumer("lateNoDLQ.0", noDLQInputChannel, properties);
+
+		MessageChannel pubSubOutputChannel = new DirectChannel();
+		binder.bindPubSubProducer("latePubSub", pubSubOutputChannel, properties);
+		QueueChannel pubSubInputChannel = new QueueChannel();
+		binder.bindPubSubConsumer("latePubSub", pubSubInputChannel, "lategroup", properties);
+		QueueChannel durablePubSubInputChannel = new QueueChannel();
+		properties.setProperty("durableSubscription", "true");
+		binder.bindPubSubConsumer("latePubSub", durablePubSubInputChannel, "lateDurableGroup", properties);
+
+		proxy.start();
+
+		moduleOutputChannel.send(new GenericMessage<>("foo"));
+		Message<?> message = moduleInputChannel.receive(10000);
+		assertNotNull(message);
+		assertEquals("foo", message.getPayload());
+
+		noDLQOutputChannel.send(new GenericMessage<>("bar"));
+		message = noDLQInputChannel.receive(10000);
+		assertNotNull(message);
+		assertEquals("bar", message.getPayload());
+
+		pubSubOutputChannel.send(new GenericMessage<>("baz"));
+		message = pubSubInputChannel.receive(10000);
+		assertNotNull(message);
+		assertEquals("baz", message.getPayload());
+		message = durablePubSubInputChannel.receive(10000);
+		assertNotNull(message);
+		assertEquals("baz", message.getPayload());
+
+		partOutputChannel.send(new GenericMessage<>("0"));
+		partOutputChannel.send(new GenericMessage<>("1"));
+		message = partInputChannel0.receive(10000);
+		assertNotNull(message);
+		assertEquals("0", message.getPayload());
+		message = partInputChannel1.receive(10000);
+		assertNotNull(message);
+		assertEquals("1", message.getPayload());
+
+		binder.unbindProducer("late.0", moduleOutputChannel);
+		binder.unbindConsumer("late.0", moduleInputChannel);
+		binder.unbindProducer("partlate.0", moduleOutputChannel);
+		binder.unbindConsumers("partlate.0");
+
+		proxy.stop();
+		cf.destroy();
+
+		RabbitAdmin admin = new RabbitAdmin(this.rabbitAvailableRule.getResource());
+		admin.deleteQueue("latebinder.late.0");
+		admin.deleteQueue("latebinder.lateNoDLQ.0");
+		admin.deleteQueue("latebinder.partlate.0-0");
+		admin.deleteQueue("latebinder.partlate.0-1");
+		admin.deleteQueue("latebinder.late.0.dlq");
+		admin.deleteQueue("latebinder.partlate.0-0.dlq");
+		admin.deleteQueue("latebinder.partlate.0-1.dlq");
+		admin.deleteQueue("latebinder.lateDurableGroup.latePubSub");
+		admin.deleteExchange("latebinder.late.0");
+		admin.deleteExchange("latebinder.lateNoDLQ.0");
+		admin.deleteExchange("latebinder.partlate.0");
+		admin.deleteExchange("latebinder.latePubSub");
+		admin.deleteExchange("latebinder.DLX");
+		this.rabbitAvailableRule.getResource().destroy();
+	}
+
 	private SimpleMessageListenerContainer verifyContainer(AbstractEndpoint endpoint) {
 		SimpleMessageListenerContainer container;
 		Advice retry;

spring-cloud-stream-test-support-internal/src/main/java/org/springframework/cloud/stream/test/junit/rabbit/RabbitTestSupport.java

@@ -16,8 +16,15 @@
 
 package org.springframework.cloud.stream.test.junit.rabbit;
 
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.ServerSocket;
 import java.net.Socket;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 
+import javax.net.ServerSocketFactory;
 import javax.net.SocketFactory;
 
 import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
@@ -58,4 +65,102 @@ public class RabbitTestSupport extends AbstractExternalResourceTestSupport<Cachi
 		resource.destroy();
 	}
 
+	/**
+	 * Test class to allow testing deferred entity declarations when RabbitMQ is down.
+	 *
+	 */
+	public static class RabbitProxy {
+
+		private final int port;
+
+		private final ExecutorService serverExec = Executors.newSingleThreadExecutor();
+
+		private final ExecutorService socketExec = Executors.newCachedThreadPool();
+
+		private volatile ServerSocket serverSocket;
+
+		public RabbitProxy() throws IOException {
+			ServerSocket serverSocket = ServerSocketFactory.getDefault().createServerSocket(0);
+			this.port = serverSocket.getLocalPort();
+			serverSocket.close();
+		}
+
+		public int getPort() {
+			return this.port;
+		}
+
+		public void start() throws IOException {
+			this.serverSocket = ServerSocketFactory.getDefault().createServerSocket(this.port);
+			this.serverExec.execute(new Runnable() {
+
+				@Override
+				public void run() {
+					try {
+						while (true) {
+							final Socket socket = serverSocket.accept();
+							socketExec.execute(new Runnable() {
+
+								@Override
+								public void run() {
+									try {
+										final Socket rabbitSocket = SocketFactory.getDefault().createSocket("localhost",
+												5672);
+										socketExec.execute(new Runnable() {
+
+											@Override
+											public void run() {
+												try {
+													InputStream is = rabbitSocket.getInputStream();
+													OutputStream os = socket.getOutputStream();
+													int c;
+													while ((c = is.read()) >= 0) {
+														os.write(c);
+													}
+												}
+												catch (IOException e) {
+													try {
+														socket.close();
+														rabbitSocket.close();
+													}
+													catch (IOException e1) {
+													}
+												}
+											}
+										});
+										InputStream is = socket.getInputStream();
+										OutputStream os = rabbitSocket.getOutputStream();
+										int c;
+										while ((c = is.read()) >= 0) {
+											os.write(c);
+										}
+									}
+									catch (IOException e) {
+										try {
+											socket.close();
+										}
+										catch (IOException e1) {
+										}
+									}
+								}
+
+							});
+						}
+					}
+					catch (IOException e) {
+						try {
+							serverSocket.close();
+						}
+						catch (IOException e1) {
+						}
+					}
+				}
+			});
+		}
+
+		public void stop() throws IOException {
+			this.serverSocket.close();
+		}
+
+	}
+
 }