@Before
	public void setUp() {
		MockitoAnnotations.initMocks(this);
		EmbeddedZooKeeper embeddedZooKeeper = new EmbeddedZooKeeper();
		embeddedZooKeeper.start();
		ZooKeeperConnection zkConnection = new ZooKeeperConnection("localhost:" + embeddedZooKeeper.getClientPort());
		zkConnection.start();
		while (!zkConnection.isConnected()) {
			try {
				Thread.sleep(100);
			}
			catch (InterruptedException e) {
				Thread.currentThread().interrupt();
				throw new IllegalStateException("interrupted while waiting for ZK connection");
			}
		}
		streamPlugin = new StreamPlugin(bus, zkConnection);
		when(module.getComponent("input", MessageChannel.class)).thenReturn(input);
		when(module.getComponent("output", MessageChannel.class)).thenReturn(output);
		when(module.getDescriptor()).thenReturn(descriptor);
		deploymentProperies = new ModuleDeploymentProperties();
		deploymentProperies.put("consumer.foo", "bar");
		deploymentProperies.put("producer.baz", "qux");
		when(module.getDeploymentProperties()).thenReturn(deploymentProperies);
		doAnswer(new Answer<Void>() {

			@Override
			public Void answer(InvocationOnMock invocation) throws Throwable {
				consumerProperties = (Properties) invocation.getArguments()[2];
				return null;
			}

		}).when(bus).bindConsumer(eq("mystream.0"), same(input), any(Properties.class));
		doAnswer(new Answer<Void>() {

			@Override
			public Void answer(InvocationOnMock invocation) throws Throwable {
				producerProperties = (Properties) invocation.getArguments()[2];
				return null;
			}

		}).when(bus).bindProducer(eq("mystream.1"), same(output), any(Properties.class));
	}
@author Jennifer Hickey
@author Gary Russell
spring-xd-dirt/src/test/java/org/springframework/xd/dirt/plugins/stream/MessageBusRegistrationTests.java