INT-267: Implement Routing Slip Pattern

JIRA: https://jira.spring.io/browse/INT-267

The implementation looks like:
* There is the `ROUTING_SLIP` header to keep the list of bean ids;
* The `ROUTING_SLIP_INDEX` header keeps track of current `index` in the `ROUTING_SLIP` header;
* `ROUTING_SLIP` List can contain channel names or bean references for the `RoutingSlip` strategy implementations.
They are differentiated with `@` prefix;
* The `<header-enricher>` adds `<routing-slip>` sub-element to specify the comma-delimited value for desired `ROUTING_SLIP` for the downstream flow;
* The `AbstractReplyProducingMessageHandler` adds the logic to get deal with `ROUTING_SLIP` List and the algorithm is:
    - If `ROUTING_SLIP` isn't `null` we build `AtomicInteger` for the current `routingSlipIndex`;
    - the recursive `getReplyChannelFromRoutingSlip` should return a channel name or `null`;
    - if current `ROUTING_SLIP_INDEX` if for the `RoutingSlip` strategy, we check its result for `null` and `incrementAndGet()` the current index or not;
    - for the simple channel name value from `ROUTING_SLIP` list we just `incrementAndGet()` the current index and return the value;
    - the new `ROUTING_SLIP_INDEX` is populated to the headers of new reply message.
* Polishing for `AbstractMessageSplitter`

**TODO**: Docs and applying `RoutingSlip` algorithm for the `AbstractCorrelatingMessageHandler`

INT-267: Move `replyProducing` logic `ARPMH` -> `AMPH`

* Rework `AbstractCorrelatingMessageHandler` to use methods from super class
* Rework `MessageHandlerChain.ReplyForwardingMessageChannel` to use `produceReply`
* Rename `RoutingSlip` -> `RoutingSlipRouteStrategy`
* Add `ExpressionEvaluationRoutingSlipRouteStrategy`

Now `routingSlip` header can be configured like:

```
<routing-slip value="channel1; #{@routingSlipRoutingPojo.get(request, reply)}; @routingSlipRoutingStrategy; #{request.headers[myRoutingSlipChannel]}; channel6"/>
```

Where `;` is used as delimiter, because of `,` in the method invocation from SpEL.
The simple literal (`channel1`) is just a `MessageChannel` `id`.
`@` is used for `RoutingSlipRouteStrategy` bean reference.
`#{...}` used for SpEL.
The `HeaderEnricherParserSupport` parses this `value` to the `List<String>` - a set of bean names, where any SpEL is wrapped
to the `ExpressionEvaluationRoutingSlipRouteStrategy` bean definition

INT-267: Introduce `RoutingSlip` Domain class

Rename `AbstractMessageProducingHandler` methods: `*reply` -> `*output`

Conflicts:
	src/reference/docbook/whats-new.xml

INT-267: Rework `RoutingSlip` -> `Map<List<String>, Integer>`

Since `RoutingSlip` POJO isn't scalable in the distributed multi-language environment,
it would be better to use some Java generic type for this `ROUTING_SLIP` header.

The `Collections.singletonMap(Collections.unmodifiableList(routingSlipPath), 0)` is the best candidate to be convertible to other systems
and allow to have thread-safety.

The `ROUTING_SLIP` header is recalculated now on each `nextPath`

INT-267: Introduce `RoutingSlipHeaderValueMessageProcessor`

* Rework `HeaderEnricherParserSupport` logic to get rid of SpEL parsing
and change `routingSlipPath` to the `ManagedList<String>` to get gain of `property-placehoder`
* Add `<context:property-placeholder>` stuff to the `RoutingSlipTests`

INT-267: Move inline expressions to the implicit `EERSRS` from the `RoutingSlipHeaderValueMessageProcessor`

INT-267: Fix up JavaDocs and add JDBC test-case

INT-267: Address PR comments

INT-267: Polishing according PR comments

Polishing

spring-integration-core/src/test/java/org/springframework/integration/config/ChainParserTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2013 the original author or authors.
+ * Copyright 2002-2014 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -278,7 +278,8 @@ public class ChainParserTests {
 
 	@Test // INT-1165
 	public void chainWithSendTimeout() {
-		long sendTimeout = TestUtils.getPropertyValue(this.chainWithSendTimeout, "sendTimeout", Long.class);
+		long sendTimeout = TestUtils.getPropertyValue(this.chainWithSendTimeout, "messagingTemplate.sendTimeout",
+				Long.class);
 		assertEquals(9876, sendTimeout);
 	}
 
@@ -343,7 +344,7 @@ public class ChainParserTests {
 		assertEquals(256, chainEndpoint.getPhase());
 
 		MessageHandlerChain handlerChain = ctx.getBean("chain.handler", MessageHandlerChain.class);
-		assertEquals(3000L, TestUtils.getPropertyValue(handlerChain, "sendTimeout"));
+		assertEquals(3000L, TestUtils.getPropertyValue(handlerChain, "messagingTemplate.sendTimeout"));
 		assertEquals(false, TestUtils.getPropertyValue(handlerChain, "running"));
 		//INT-3108
 		MessageHandler serviceActivator = ctx.getBean("chain$child.sa-within-chain.handler", MessageHandler.class);

spring-integration-core/src/test/java/org/springframework/integration/config/xml/HeaderEnricherTests-context.xml

@@ -76,6 +76,16 @@
 		<priority expression="payload.priority"/>
 	</header-enricher>
 
+	<beans:bean id="bazRoutingSlip" class="org.mockito.Mockito" factory-method="mock">
+		<beans:constructor-arg value="org.springframework.integration.routingslip.RoutingSlipRouteStrategy"/>
+	</beans:bean>
+
+	<channel id="fooChannel"/>
+
+	<header-enricher input-channel="routingSlipInput">
+		<routing-slip value="fooChannel; barExpression; bazRoutingSlip"/>
+	</header-enricher>
+
 	<header-enricher input-channel="payloadExpressionInput">
 		<header name="testHeader" expression="payload.name + 'bar'"/>
 	</header-enricher>

spring-integration-core/src/test/java/org/springframework/integration/config/xml/HeaderEnricherTests.java

@@ -16,13 +16,16 @@
 
 package org.springframework.integration.config.xml;
 
+import static org.hamcrest.Matchers.instanceOf;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
 
 import java.util.Collections;
 import java.util.Date;
+import java.util.List;
 import java.util.Map;
 
 import org.junit.Test;
@@ -34,6 +37,7 @@ import org.springframework.context.ApplicationContext;
 import org.springframework.context.support.ClassPathXmlApplicationContext;
 import org.springframework.integration.IntegrationMessageHeaderAccessor;
 import org.springframework.integration.core.MessagingTemplate;
+import org.springframework.integration.routingslip.ExpressionEvaluatingRoutingSlipRouteStrategy;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.integration.transformer.MessageTransformationException;
 import org.springframework.messaging.Message;
@@ -119,7 +123,7 @@ public class HeaderEnricherTests {
 		assertNotNull(result);
 		Object correlationId = new IntegrationMessageHeaderAccessor(result).getCorrelationId();
 		assertEquals(Long.class, correlationId.getClass());
-		assertEquals(new Long(123), correlationId);
+		assertEquals(123L, correlationId);
 	}
 
 	@Test
@@ -128,7 +132,7 @@ public class HeaderEnricherTests {
 		MessageChannel channel = context.getBean("correlationIdRefInput", MessageChannel.class);
 		Message<?> result = template.sendAndReceive(channel, new GenericMessage<String>("test"));
 		assertNotNull(result);
-		assertEquals(new Integer(123), new IntegrationMessageHeaderAccessor(result).getCorrelationId());
+		assertEquals(123, new IntegrationMessageHeaderAccessor(result).getCorrelationId());
 	}
 
 	@Test
@@ -162,7 +166,8 @@ public class HeaderEnricherTests {
 	public void priorityExpression() {
 		MessagingTemplate template = new MessagingTemplate();
 		MessageChannel channel = context.getBean("priorityExpressionInput", MessageChannel.class);
-		Message<?> result = template.sendAndReceive(channel, new GenericMessage<Map<String, String>>(Collections.singletonMap("priority", "-10")));
+		Message<?> result = template.sendAndReceive(channel,
+				new GenericMessage<Map<String, String>>(Collections.singletonMap("priority", "-10")));
 		assertNotNull(result);
 		assertEquals(new Integer(-10), new IntegrationMessageHeaderAccessor(result).getPriority());
 	}
@@ -205,7 +210,7 @@ public class HeaderEnricherTests {
 		Message<?> result = template.sendAndReceive(channel, new GenericMessage<String>("test"));
 		assertNotNull(result);
 		assertEquals(Long.class, result.getHeaders().get("number").getClass());
-		assertEquals(new Long(12345), result.getHeaders().get("number"));
+		assertEquals(12345L, result.getHeaders().get("number"));
 	}
 
 	@Test
@@ -252,9 +257,29 @@ public class HeaderEnricherTests {
 
 	@Test(expected = BeanDefinitionParsingException.class)
 	public void testFailConfigUnexpectedSubElement() {
-		new ClassPathXmlApplicationContext("HeaderEnricherWithUnexpectedSubElementForHeader-fail-context.xml", this.getClass());
+		new ClassPathXmlApplicationContext("HeaderEnricherWithUnexpectedSubElementForHeader-fail-context.xml",
+				this.getClass());
 	}
 
+	@Test
+	public void testRoutingSlip() {
+		MessagingTemplate template = new MessagingTemplate();
+		MessageChannel channel = context.getBean("routingSlipInput", MessageChannel.class);
+		Message<?> result = template.sendAndReceive(channel, new GenericMessage<String>("test"));
+		assertNotNull(result);
+		Object routingSlip = new IntegrationMessageHeaderAccessor(result)
+				.getHeader(IntegrationMessageHeaderAccessor.ROUTING_SLIP);
+		assertNotNull(routingSlip);
+		assertThat(routingSlip, instanceOf(Map.class));
+		@SuppressWarnings("unchecked")
+		List<Object> routingSlipPath = (List<Object>) ((Map) routingSlip).keySet().iterator().next();
+
+		assertEquals("fooChannel", routingSlipPath.get(0));
+		assertThat(routingSlipPath.get(1), instanceOf(ExpressionEvaluatingRoutingSlipRouteStrategy.class));
+		assertEquals("bazRoutingSlip", routingSlipPath.get(2));
+	}
+
+
 	public static class TestBean {
 
 		private final String name;
@@ -274,9 +299,8 @@ public class HeaderEnricherTests {
 
 			TestBean testBean = (TestBean) o;
 
-			if (name != null ? !name.equals(testBean.name) : testBean.name != null) return false;
+			return !(name != null ? !name.equals(testBean.name) : testBean.name != null);
 
-			return true;
 		}
 
 		@Override

spring-integration-core/src/test/java/org/springframework/integration/handler/MessageHandlerChainTests.java

@@ -86,7 +86,7 @@ public class MessageHandlerChainTests {
 		chain.setHandlers(handlers);
 		chain.setOutputChannel(outputChannel);
 		chain.handleMessage(message);
-		Mockito.verify(outputChannel).send(Mockito.eq(message), Mockito.eq(-1L));
+		Mockito.verify(outputChannel).send(Mockito.eq(message));
 	}
 
 	@Test(expected = IllegalArgumentException.class)

spring-integration-core/src/test/java/org/springframework/integration/routingslip/RoutingSlipTests-context.xml

@@ -0,0 +1,64 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans:beans xmlns="http://www.springframework.org/schema/integration"
+			 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+			 xmlns:beans="http://www.springframework.org/schema/beans"
+			 xmlns:task="http://www.springframework.org/schema/task"
+			 xmlns:context="http://www.springframework.org/schema/context"
+			 xmlns:util="http://www.springframework.org/schema/util"
+			 xsi:schemaLocation="http://www.springframework.org/schema/beans
+			http://www.springframework.org/schema/beans/spring-beans.xsd
+			http://www.springframework.org/schema/integration
+			http://www.springframework.org/schema/integration/spring-integration.xsd
+			http://www.springframework.org/schema/task
+			http://www.springframework.org/schema/task/spring-task.xsd
+			http://www.springframework.org/schema/context
+			http://www.springframework.org/schema/context/spring-context.xsd
+			http://www.springframework.org/schema/util
+			http://www.springframework.org/schema/util/spring-util.xsd">
+
+	<util:properties id="properties">
+		<beans:prop key="myRoutePath1">channel1</beans:prop>
+		<beans:prop key="myRoutePath2">request.headers[myRoutingSlipChannel]</beans:prop>
+	</util:properties>
+
+	<context:property-placeholder properties-ref="properties"/>
+
+	<message-history/>
+
+	<task:executor id="executor"/>
+
+	<beans:bean id="routingSlipRoutingPojo"
+				class="org.springframework.integration.routingslip.RoutingSlipTests$TestRoutingSlipRoutePojo"/>
+
+	<beans:bean id="routingSlipRoutingStrategy"
+				class="org.springframework.integration.routingslip.RoutingSlipTests$TestRoutingSlipRouteStrategy"/>
+
+	<header-enricher input-channel="input" output-channel="split">
+		<routing-slip
+				value="${myRoutePath1}; @routingSlipRoutingPojo.get(request, reply);
+							routingSlipRoutingStrategy; ${myRoutePath2}; aggregate"/>
+	</header-enricher>
+
+	<splitter input-channel="split" output-channel="process"/>
+
+	<channel id="process">
+		<dispatcher task-executor="executor"/>
+	</channel>
+
+	<bridge input-channel="process"/>
+
+	<bridge input-channel="channel1"/>
+
+	<bridge input-channel="channel2"/>
+
+	<bridge input-channel="channel3"/>
+
+	<bridge input-channel="channel4"/>
+
+	<chain input-channel="channel5">
+		<header-filter header-names="myRoutingSlipChannel"/>
+	</chain>
+
+	<aggregator input-channel="aggregate" expression="new java.util.ArrayList(#root)"/>
+
+</beans:beans>

spring-integration-core/src/test/java/org/springframework/integration/routingslip/RoutingSlipTests.java

@@ -0,0 +1,107 @@
+/*
+ * Copyright 2014 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.routingslip;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.integration.IntegrationMessageHeaderAccessor;
+import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.history.MessageHistory;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.PollableChannel;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+/**
+ * @author Artem Bilan
+ * @since 4.1
+ */
+@ContextConfiguration
+@RunWith(SpringJUnit4ClassRunner.class)
+public class RoutingSlipTests {
+
+	@Autowired
+	private MessageChannel input;
+
+	@Test
+	@SuppressWarnings("unchecked")
+	public void testRoutingSlip() {
+		PollableChannel replyChannel = new QueueChannel();
+		Message<List<String>> request = MessageBuilder.withPayload(Arrays.asList("test1", "test2"))
+				.setReplyChannel(replyChannel)
+				.setHeader("myRoutingSlipChannel", "channel5").build();
+		this.input.send(request);
+		Message<?> reply = replyChannel.receive(10000);
+		assertNotNull(reply);
+		List<Message<?>> messages = (List<Message<?>>) reply.getPayload();
+		for (Message<?> message : messages) {
+			Map<List<String>, Integer> routingSlip = message.getHeaders()
+					.get(IntegrationMessageHeaderAccessor.ROUTING_SLIP, Map.class);
+			assertEquals(routingSlip.keySet().iterator().next().size(), routingSlip.values().iterator().next().intValue());
+			MessageHistory messageHistory = MessageHistory.read(message);
+			List<String> channelNames = Arrays.asList("input", "split", "process", "channel1", "channel2",
+					"channel3", "channel4", "channel5", "aggregate");
+			int i = 0;
+			for (Properties properties : messageHistory) {
+				assertTrue(channelNames.contains(properties.getProperty("name")));
+			}
+		}
+	}
+
+	public static class TestRoutingSlipRoutePojo {
+
+		final String[] channels = {"channel2", "channel3"};
+
+		private int i = 0;
+
+		public String get(Message<?> requestMessage, Object reply) {
+			try {
+				return this.channels[i++];
+			}
+			catch (Exception e) {
+				return null;
+			}
+		}
+
+	}
+
+	public static class TestRoutingSlipRouteStrategy implements RoutingSlipRouteStrategy {
+
+		private AtomicBoolean invoked = new AtomicBoolean();
+
+		@Override
+		public String getNextPath(Message<?> requestMessage, Object reply) {
+			return !invoked.getAndSet(true) ? "channel4" : null;
+		}
+
+	}
+
+}