private void tapTest(StreamDefinition streamDefinition, StreamDefinition tapDefinition) {
		integrationSupport.createAndDeployStream(streamDefinition);
		integrationSupport.createAndDeployStream(tapDefinition);

		MessageBus bus = integrationSupport.messageBus();

		NamedChannelSource source = new SingleNodeNamedChannelSourceFactory(bus).createNamedChannelSource("queue:source");
		NamedChannelSink streamSink = new SingleNodeNamedChannelSinkFactory(bus).createNamedChannelSink("queue:sink");
		NamedChannelSink tapSink = new SingleNodeNamedChannelSinkFactory(bus).createNamedChannelSink("queue:tap");

		source.send(new GenericMessage<String>("Dracarys!"));

		Message<?> m1;
		int count1 = 0;
		String result1 = null;
		while ((m1 = streamSink.receive(1000)) != null) {
			count1++;
			result1 = (String) m1.getPayload();
		}

		Message<?> m2;
		int count2 = 0;
		String result2 = null;
		while ((m2 = tapSink.receive(1000)) != null) {
			count2++;
			result2 = (String) m2.getPayload();
		}

		assertEquals("DRACARYS!", result1);
		assertEquals(1, count1);

		assertEquals("DR.C.RYS!", result2);
		assertEquals(1, count2);

		source.unbind();
		streamSink.unbind();
		tapSink.unbind();
		integrationSupport.undeployAndDestroyStream(streamDefinition);
		integrationSupport.undeployAndDestroyStream(tapDefinition);
	}
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