@Test
	public void mqttSourceStreamReceivesMqttSinkStreamOutputBinary() throws Exception {
		StreamDefinition mqtt3 = new StreamDefinition("mqtt3", "queue:mqttsource2 > mqtt --topic=foo2");
		StreamDefinition mqtt4 = new StreamDefinition("mqtt4", "mqtt --topics=foo2 --binary=true > queue:mqttsink2");
		integrationSupport.createAndDeployStream(mqtt3);
		integrationSupport.createAndDeployStream(mqtt4);

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

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

		assertThat(result, instanceOf(byte[].class));
		assertEquals("hello", new String((byte[]) result));
		source.unbind();
		sink.unbind();
	}
@author Mark Fisher
@author Gary Russell
spring-xd-dirt/src/test/java/org/springframework/xd/dirt/stream/RabbitSingleNodeStreamDeploymentIntegrationTests.java