@Test
	public void testTapDoesntHurtStream() throws Exception {
		LocalMessageBus bus = (LocalMessageBus) getMessageBus();
		DirectChannel moduleOutputChannel = new DirectChannel();
		moduleOutputChannel.setBeanName("bangOut");
		DirectChannel tapChannel = new DirectChannel();
		tapChannel.setBeanName("tapChannel");
		WireTap tap = new WireTap(tapChannel);
		moduleOutputChannel.addInterceptor(tap);
		bus.bindProducer("bang.0", moduleOutputChannel, null);
		final AtomicBoolean messageReceived = new AtomicBoolean();
		final AtomicReference<Thread> streamThread = new AtomicReference<Thread>();
		bus.bindConsumer("bang.0", new DirectChannel() {

			@Override
			protected boolean doSend(Message<?> message, long timeout) {
				messageReceived.set(true);
				streamThread.set(Thread.currentThread());
				return true;
			}
		}, null);
		final CountDownLatch tapped = new CountDownLatch(1);
		final AtomicReference<Thread> tapThread = new AtomicReference<Thread>();
		bus.bindPubSubProducer("tap:stream:bang.0", tapChannel, null);
		bus.bindPubSubConsumer("tap:stream:bang.0", new DirectChannel() {

			@Override
			protected boolean doSend(Message<?> message, long timeout) {
				tapThread.set(Thread.currentThread());
				tapped.countDown();
				throw new RuntimeException("bang");
			}
		}, null);
		moduleOutputChannel.send(new GenericMessage<String>("Foo"));
		assertTrue(tapped.await(10, TimeUnit.SECONDS));
		assertTrue(messageReceived.get());
		assertSame(Thread.currentThread(), streamThread.get());
		assertNotNull(tapThread.get());
		assertNotSame(Thread.currentThread(), tapThread.get());
	}
@author Gary Russell
@author David Turanski
@since 1.0
spring-xd-dirt/src/test/java/org/springframework/xd/dirt/integration/bus/local/LocalMessageBusTests.java