@Test
	public void chainWithSink() {

		String processingChainUnderTest = "transform --expression='T(org.joda.time.format.DateTimeFormat).forPattern(\"yyyy-MM-dd HH:mm:ss\").parseDateTime(payload)' | log";

		SingleNodeProcessingChainProducer chain = chainProducer(application, "dateToDateTime", processingChainUnderTest);

		StreamDefinition tap = new StreamDefinition("testtap", "tap:stream:dateToDateTime.0 > queue:tap");
		integrationSupport.createAndDeployStream(tap);
		NamedChannelSink sink = new SingleNodeNamedChannelSinkFactory(integrationSupport.messageBus()).createNamedChannelSink("queue:tap");

		chain.sendPayload(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));

		Object payload = sink.receivePayload(RECEIVE_TIMEOUT);
		assertNotNull(payload);
		assertTrue(payload instanceof DateTime);

		integrationSupport.undeployAndDestroyStream(tap);
		chain.destroy();
	}
Test a processing chain that provides a sink. Send a payload and set up a tap to verify the result;
spring-xd-dirt/src/test/java/org/springframework/xd/dirt/stream/ExampleProcessingChainTests.java