@Test
	public void testDurablePubSubWithAutoBindDLQ() throws Exception {
		RabbitAdmin admin = new RabbitAdmin(this.rabbitAvailableRule.getResource());

		MessageBus bus = getMessageBus();
		Properties properties = new Properties();
		properties.put("prefix", "xdbustest.");
		properties.put("autoBindDLQ", "true");
		properties.put("durableSubscription", "true");
		properties.put("maxAttempts", "1"); // disable retry
		properties.put("requeue", "false");
		properties.put("concurrency", "2");
		DirectChannel moduleInputChannel = new DirectChannel();
		moduleInputChannel.setBeanName("durableTest");
		final List<Thread> threads = new ArrayList<>();
		moduleInputChannel.subscribe(new MessageHandler() {

			@Override
			public void handleMessage(Message<?> message) throws MessagingException {
				threads.add(Thread.currentThread());
				throw new RuntimeException("foo");
			}

		});
		bus.bindPubSubConsumer("teststream.tap:stream:durabletest.0", moduleInputChannel, properties);

		RabbitTemplate template = new RabbitTemplate(this.rabbitAvailableRule.getResource());
		template.convertAndSend("xdbustest.topic.tap:stream:durabletest.0", "", "foo");

		template.setReceiveTimeout(10000);
		Object deadLetter = template.receiveAndConvert("xdbustest.teststream.tap:stream:durabletest.0.dlq");
		assertNotNull(deadLetter);
		assertEquals("foo", deadLetter);

		template.convertAndSend("xdbustest.topic.tap:stream:durabletest.0", "", "bar");

		deadLetter = template.receiveAndConvert("xdbustest.teststream.tap:stream:durabletest.0.dlq");
		assertNotNull(deadLetter);
		assertEquals("bar", deadLetter);

		template.setReceiveTimeout(10);
		assertNull(template.receiveAndConvert("xdbustest.teststream.tap:stream:durabletest.0.dlq"));
		assertEquals(2, threads.size());
		assertNotSame(threads.get(0), threads.get(1));

		@SuppressWarnings("unchecked")
		List<Binding> bindings = TestUtils.getPropertyValue(bus, "messageBus.bindings", List.class);
		assertEquals(1, bindings.size());
		assertEquals(2,
				TestUtils.getPropertyValue(bindings.get(0), "endpoint.messageListenerContainer.concurrentConsumers"));

		bus.unbindConsumer("teststream.tap:stream:durabletest.0", moduleInputChannel);
		assertNotNull(admin.getQueueProperties("xdbustest.teststream.tap:stream:durabletest.0.dlq"));
		admin.deleteQueue("xdbustest.teststream.tap:stream:durabletest.0.dlq");
		admin.deleteQueue("xdbustest.teststream.tap:stream:durabletest.0");
		admin.deleteExchange("xdbustest.topic.tap:stream:durabletest.0");
		admin.deleteExchange("xdbustest.DLX");

		assertEquals(0, bindings.size());
	}
@author Mark Fisher
@author Gary Russell
spring-xd-dirt/src/test/java/org/springframework/xd/dirt/integration/bus/rabbit/RabbitMessageBusTests.java