@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();
}