@Test
	public void verifyTopicChannelsRegisteredOnDemand() throws InterruptedException {
		final StreamDefinition routerDefinition = new StreamDefinition("routerDefinition",
				"topic:x > router --expression=payload.contains('y')?'topic:y':'topic:z'");
		integrationSupport.streamDefinitionRepository().save(routerDefinition);
		integrationSupport.deployStream(routerDefinition, onDemandProperties());
		Thread.sleep(1000);

		MessageBus bus = testMessageBus != null ? testMessageBus : integrationSupport.messageBus();
		if (bus instanceof LocalMessageBus) { // no such bean for other buses
			singleNodeApplication.pluginContext().getBean("topic:x", MessageChannel.class);
		}
		assertFalse(singleNodeApplication.pluginContext().containsBean("topic:y"));
		assertFalse(singleNodeApplication.pluginContext().containsBean("topic:z"));


		DirectChannel testChannel = new DirectChannel();
		bus.bindPubSubProducer("topic:x", testChannel, null);
		testChannel.send(MessageBuilder.withPayload("y").build());
		Thread.sleep(2000);

		singleNodeApplication.pluginContext().getBean("topic:y", MessageChannel.class);
		assertFalse(singleNodeApplication.pluginContext().containsBean("topic:z"));

		testChannel.send(MessageBuilder.withPayload("z").build());
		Thread.sleep(2000);
		MessageChannel y3 = singleNodeApplication.pluginContext().getBean("topic:y", MessageChannel.class);
		MessageChannel z3 = singleNodeApplication.pluginContext().getBean("topic:z", MessageChannel.class);
		assertNotNull(y3);
		assertNotNull(z3);

		QueueChannel consumer = new QueueChannel();
		bus.bindPubSubConsumer("topic:y", consumer, null);
		bus.bindPubSubConsumer("topic:z", consumer, null);
		testChannel.send(MessageBuilder.withPayload("y").build());
		Thread.sleep(2000);
		testChannel.send(MessageBuilder.withPayload("z").build());
		Thread.sleep(2000);
		assertEquals("y", consumer.receive(2000).getPayload());
		assertEquals("z", consumer.receive(2000).getPayload());
		assertEquals(0, consumer.getQueueSize());

		verifyDynamicProperties(bus, "topic");

		bus.unbindProducer("topic:x", testChannel);
		bus.unbindConsumers("topic:x");
		bus.unbindConsumers("topic:y");
		bus.unbindConsumers("topic:z");
		bus.unbindProducers("topic:y");
		bus.unbindProducers("topic:z");
	}
Verifies the content of the queues

@param y3 the channel where the y's are sent
@param z3 the channel where the z's are sent
@param initialTransportState the initial state of the transport (can be ignored except for Kafka)
spring-xd-dirt/src/test/java/org/springframework/xd/dirt/stream/AbstractSingleNodeStreamDeploymentIntegrationTests.java