@Test
	public void testDirectBinding() throws Exception {
		MessageBus bus = getMessageBus();
		Properties properties = new Properties();
		properties.setProperty(BusProperties.DIRECT_BINDING_ALLOWED, "true");

		DirectChannel moduleInputChannel = new DirectChannel();
		moduleInputChannel.setBeanName("direct.input");
		DirectChannel moduleOutputChannel = new DirectChannel();
		moduleOutputChannel.setBeanName("direct.output");
		bus.bindConsumer("direct.0", moduleInputChannel, null);
		bus.bindProducer("direct.0", moduleOutputChannel, properties);

		final AtomicReference<Thread> caller = new AtomicReference<Thread>();
		final AtomicInteger count = new AtomicInteger();
		moduleInputChannel.subscribe(new MessageHandler() {

			@Override
			public void handleMessage(Message<?> message) throws MessagingException {
				caller.set(Thread.currentThread());
				count.incrementAndGet();
			}
		});

		moduleOutputChannel.send(new GenericMessage<String>("foo"));
		moduleOutputChannel.send(new GenericMessage<String>("foo"));

		assertNotNull(caller.get());
		assertSame(Thread.currentThread(), caller.get());
		assertEquals(2, count.get());
		assertNull(spyOn("direct.0").receive(true));

		// Remove direct binding and bind producer to the bus
		bus.unbindConsumers("direct.0");
		busBindUnbindLatency();

		Spy spy = spyOn("direct.0");
		count.set(0);
		moduleOutputChannel.send(new GenericMessage<String>("bar"));
		moduleOutputChannel.send(new GenericMessage<String>("baz"));
		Object bar = spy.receive(false);
		assertEquals("bar", bar);
		Object baz = spy.receive(false);
		assertEquals("baz", baz);
		assertEquals(0, count.get());

		// Unbind producer from bus and bind directly again
		caller.set(null);
		bus.bindConsumer("direct.0", moduleInputChannel, null);
		moduleOutputChannel.send(new GenericMessage<String>("foo"));
		moduleOutputChannel.send(new GenericMessage<String>("foo"));
		assertNotNull(caller.get());
		assertSame(Thread.currentThread(), caller.get());
		assertEquals(2, count.get());
		assertNull(spy.receive(true));

		bus.unbindProducers("direct.0");
		bus.unbindConsumers("direct.0");
	}
Tests for buses that use an external broker.

@author Gary Russell
spring-xd-dirt/src/test/java/org/springframework/xd/dirt/integration/bus/BrokerBusTests.java