private void doTest(StreamDefinition routerDefinition) throws InterruptedException {
		assertEquals(0, integrationSupport.streamRepository().count());
		integrationSupport.streamDeployer().save(routerDefinition);
		assertTrue("stream not deployed", integrationSupport.deployStream(routerDefinition));
		assertEquals(1, integrationSupport.streamRepository().count());

		Thread.sleep(1000);
		assertModuleRequest(routerDefinition.getName(), "router", false);

		MessageBus bus = testMessageBus != null ? testMessageBus : integrationSupport.messageBus();
		assertNotNull(bus);

		SingleNodeNamedChannelSinkFactory sinkFactory = new SingleNodeNamedChannelSinkFactory(bus);

		NamedChannelSink foosink = sinkFactory.createNamedChannelSink(QUEUE_FOO);
		NamedChannelSink barsink = sinkFactory.createNamedChannelSink(QUEUE_BAR);

		NamedChannelSource source = new SingleNodeNamedChannelSourceFactory(bus).createNamedChannelSource(QUEUE_ROUTE);

		source.sendPayload("a");
		source.sendPayload("b");

		final Object fooPayload = foosink.receivePayload(10000);
		final Object barPayload = barsink.receivePayload(10000);

		assertEquals("a", fooPayload);
		assertEquals("b", barPayload);

		source.unbind();
		foosink.unbind();
		barsink.unbind();
	}
Verifies the content of the queues

@param y3 the channel where the y's are sent
@param z3 the channel where the z's are sent
@param initialTransportState the initial state of the transport (can be ignored except for Kafka)
spring-xd-dirt/src/test/java/org/springframework/xd/dirt/stream/AbstractSingleNodeStreamDeploymentIntegrationTests.java