@Test
	public void mqttSourceStreamReceivesMqttSinkStreamOutput() throws Exception {
		StreamDefinition mqtt1 = new StreamDefinition("mqtt1", "queue:mqttsource > mqtt --topic=foo --async=true");
		StreamDefinition mqtt2 = new StreamDefinition("mqtt2", "mqtt --topics=foo --charset=UTF-8 > queue:mqttsink");
		integrationSupport.createAndDeployStream(mqtt1);
		integrationSupport.createAndDeployStream(mqtt2);

		NamedChannelSource source = new SingleNodeNamedChannelSourceFactory(integrationSupport.messageBus()).createNamedChannelSource("queue:mqttsource");
		NamedChannelSink sink = new SingleNodeNamedChannelSinkFactory(integrationSupport.messageBus()).createNamedChannelSink("queue:mqttsink");

		Thread.sleep(1000);
		source.sendPayload("hello");
		Object result = sink.receivePayload(1000);

		assertEquals("hello", result);
		source.unbind();
		sink.unbind();
	}
@author Mark Fisher
@author Gary Russell
spring-xd-dirt/src/test/java/org/springframework/xd/dirt/stream/RabbitSingleNodeStreamDeploymentIntegrationTests.java