@Test
	public void verifyQueueChannelsRegisteredOnDemand() throws InterruptedException {
		final StreamDefinition routerDefinition = new StreamDefinition("routerDefinition",
				"queue:x > router --expression=payload.contains('y')?'queue:y':'queue: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("queue:x", MessageChannel.class);
		}
		assertFalse(singleNodeApplication.pluginContext().containsBean("queue:y"));
		assertFalse(singleNodeApplication.pluginContext().containsBean("queue:z"));

		Map<String, Object> initialQueueState = readInitialQueueState("queue:y", "queue:z");

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

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

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

		verifyDynamicProperties(bus, "queue");

		verifyOnDemandQueues(y3, z3, initialQueueState);

		bus.unbindProducer("queue:x", testChannel);
		bus.unbindConsumer("queue:y", y3);
		bus.unbindConsumer("queue:z", z3);
	}
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