@Test
	public final void testTopicChannel() throws InterruptedException {

		StreamDefinition bar1Definition = new StreamDefinition("bar1Definition",
				"topic:foo > queue:bar1");
		StreamDefinition bar2Definition = new StreamDefinition("bar2Definition",
				"topic:foo > queue:bar2");
		assertEquals(0, integrationSupport.streamRepository().count());
		integrationSupport.streamDeployer().save(bar1Definition);
		integrationSupport.deployStream(bar1Definition,
				Collections.singletonMap("module.*." + ModuleDeploymentProperties.TRACK_HISTORY_KEY, "true"));

		integrationSupport.streamDeployer().save(bar2Definition);
		integrationSupport.deployStream(bar2Definition);
		Thread.sleep(1000);
		assertEquals(2, integrationSupport.streamRepository().count());

		MessageBus bus = integrationSupport.messageBus();

		SingleNodeNamedChannelSinkFactory sinkFactory = new SingleNodeNamedChannelSinkFactory(bus);

		NamedChannelSink bar1sink = sinkFactory.createNamedChannelSink("queue:bar1");
		NamedChannelSink bar2sink = sinkFactory.createNamedChannelSink("queue:bar2");
		NamedChannelSource source = new SingleNodeNamedChannelSourceFactory(bus).createNamedChannelSource(TOPIC_FOO);

		source.sendPayload("hello");

		final Message<Object> bar1Message = (Message<Object>) bar1sink.receive(10000);
		assertNotNull(bar1Message);
		Object historyObject = bar1Message.getHeaders().get(XdHeaders.XD_HISTORY);
		assertNotNull(historyObject);
		assertThat(historyObject, instanceOf(List.class));
		List<Map<String, Object>> history = (List<Map<String, Object>>) historyObject;
		assertThat(history.size(), greaterThan(0));
		assertNotNull(history.get(0).get("timestamp"));
		final Object bar1 = bar1Message.getPayload();
		final Message<Object> bar2Message = (Message<Object>) bar2sink.receive(10000);
		assertNotNull(bar2Message);
		assertNull(bar2Message.getHeaders().get(XdHeaders.XD_HISTORY));
		final Object bar2 = bar2Message.getPayload();
		assertEquals("hello", bar1);
		assertEquals("hello", bar2);

		source.unbind();
		bar1sink.unbind();
		bar2sink.unbind();
	}
Get the stream 'http | log' with random available http port selected.

@return the stream
spring-xd-dirt/src/test/java/org/springframework/xd/dirt/stream/AbstractSingleNodeStreamDeploymentIntegrationTests.java