@Test
	public void testPartitionedModuleSpEL() throws Exception {
		MessageBus bus = getMessageBus();
		Properties properties = new Properties();
		properties.put("partitionKeyExpression", "payload");
		properties.put("partitionSelectorExpression", "hashCode()");
		properties.put(BusProperties.NEXT_MODULE_COUNT, "3");
		properties.put(BusProperties.NEXT_MODULE_CONCURRENCY, "2");

		DirectChannel output = new DirectChannel();
		output.setBeanName("test.output");
		bus.bindProducer("part.0", output, properties);
		@SuppressWarnings("unchecked")
		List<Binding> bindings = TestUtils.getPropertyValue(bus, "messageBus.bindings", List.class);
		assertEquals(1, bindings.size());
		try {
			AbstractEndpoint endpoint = bindings.get(0).getEndpoint();
			assertThat(getEndpointRouting(endpoint), containsString("part.0-' + headers['partition']"));
		}
		catch (UnsupportedOperationException ignored) {

		}

		properties.clear();
		properties.put("concurrency", "2");
		properties.put("partitionIndex", "0");
		properties.put("count","3");
		QueueChannel input0 = new QueueChannel();
		input0.setBeanName("test.input0S");
		bus.bindConsumer("part.0", input0, properties);
		properties.put("partitionIndex", "1");
		QueueChannel input1 = new QueueChannel();
		input1.setBeanName("test.input1S");
		bus.bindConsumer("part.0", input1, properties);
		properties.put("partitionIndex", "2");
		QueueChannel input2 = new QueueChannel();
		input2.setBeanName("test.input2S");
		bus.bindConsumer("part.0", input2, properties);

		Message<Integer> message2 = MessageBuilder.withPayload(2)
				.setHeader(IntegrationMessageHeaderAccessor.CORRELATION_ID, "foo")
				.setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER, 42)
				.setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE, 43)
				.setHeader("xdReplyChannel", "bar")
				.build();
		output.send(message2);
		output.send(new GenericMessage<Integer>(1));
		output.send(new GenericMessage<Integer>(0));

		Message<?> receive0 = input0.receive(1000);
		assertNotNull(receive0);
		Message<?> receive1 = input1.receive(1000);
		assertNotNull(receive1);
		Message<?> receive2 = input2.receive(1000);
		assertNotNull(receive2);

		Matcher<Message<?>> fooMatcher = new CustomMatcher<Message<?>>("the message with 'foo' as its correlationId") {

			@Override
			public boolean matches(Object item) {
				IntegrationMessageHeaderAccessor accessor = new IntegrationMessageHeaderAccessor((Message<?>) item);
				boolean result = "foo".equals(accessor.getCorrelationId()) &&
						42 == accessor.getSequenceNumber() &&
						43 == accessor.getSequenceSize() &&
						"bar".equals(accessor.getHeader("xdReplyChannel"));
				return result;
			}
		};

		if (usesExplicitRouting()) {
			assertEquals(0, receive0.getPayload());
			assertEquals(1, receive1.getPayload());
			assertEquals(2, receive2.getPayload());

			assertThat(receive2, fooMatcher);

		}
		else {

			assertThat(Arrays.asList(
					(Integer) receive0.getPayload(),
					(Integer) receive1.getPayload(),
					(Integer) receive2.getPayload()),
					containsInAnyOrder(0, 1, 2));

			@SuppressWarnings("unchecked")
			Matcher<Iterable<? extends Message<?>>> containsOur3Messages = containsInAnyOrder(
					fooMatcher,
					hasProperty("payload", equalTo(0)),
					hasProperty("payload", equalTo(1))
					);
			assertThat(
					Arrays.asList(receive0, receive1, receive2),
					containsOur3Messages);

		}

		bus.unbindConsumers("part.0");
		bus.unbindProducers("part.0");
	}
Tests for buses that support partitioning.

@author Gary Russell
spring-xd-dirt/src/test/java/org/springframework/xd/dirt/integration/bus/PartitionCapableBusTests.java