@Test
	public void directBindingEnabledForPartOfStream() throws InterruptedException {
		String streamName = "directBindingEnabledForPartOfStream";
		StreamDefinition sd = new StreamDefinition(streamName, String.format("http --port=%s | filter | log",
				SocketUtils.findAvailableServerSocket()));
		integrationSupport.streamDefinitionRepository().save(sd);
		Map<String, String> props = new HashMap<String, String>();
		props.put("module.http.count", "0");
		props.put("module.filter.count", "0");
		props.put("module.log.count", "1");
		integrationSupport.deployStream(sd, props);
		List<Binding> bindings = getMessageBusBindingsForStream(streamName);
		assertEquals(4, bindings.size());
		Binding logConsumerBinding = bindings.get(0);
		Binding filterProducerBinding = bindings.get(1);
		Binding filterConsumerBinding = bindings.get(2);
		Binding httpProducerBinding = bindings.get(3);
		assertEquals("inbound." + streamName + ".1", logConsumerBinding.getEndpoint().getComponentName());
		assertEquals("consumer", logConsumerBinding.getType());
		assertEquals("inbound." + streamName + ".0", filterConsumerBinding.getEndpoint().getComponentName());
		assertEquals("consumer", filterConsumerBinding.getType());
		assertEquals("outbound." + streamName + ".1", filterProducerBinding.getEndpoint().getComponentName());
		assertEquals("producer", filterProducerBinding.getType());
		assertEquals("outbound." + streamName + ".0", httpProducerBinding.getEndpoint().getComponentName());
		assertEquals("direct", httpProducerBinding.getType());
	}
Base class for testing non-local transports, such as RabbitMQ and Redis.

@author Mark Fisher
@author Gary Russell
spring-xd-dirt/src/test/java/org/springframework/xd/dirt/stream/AbstractDistributedTransportSingleNodeStreamDeploymentIntegrationTests.java