Add JMS Aggregator Sample

Remove useless `correlation-id` header mapping

basic/jms/README.md

@@ -12,6 +12,7 @@ It also uses the following components:
 1. Poller
 2. Stdout Channel Adapter (from Stream support Module)
 3. Stdin Channel Adapter (from Stream support Module)
+4. Aggregator
 
 It also shows an example of using Spring profiles to modify the configuration for test cases.
 
@@ -22,10 +23,11 @@ To run the sample, simply execute the **Main** class located in the the *org.spr
 
        $ gradlew :jms:run
 
-You will then be prompted to run one of two demos:
+You will then be prompted to run one of three demos:
 
 * **GatewayDemo**
 * **ChannelAdapterDemo**
+* **AggregationDemo**
 
 The console output should look like:
 
@@ -43,13 +45,17 @@ The console output should look like:
 
 		1. Channel Adapter Demo
 		2. Gateway Demo
+		3. Aggregation Demo
 
-When running either one of the demos you will see the following prompt:
+
+When running any of the demos you will see the following prompt:
 
 	> Please type something and hit <enter>
 
 * **GatewayDemo** uses the *DemoBean* service, which will echo the response and upper-casing it.
 * **ChannelAdapterDemo** will simply echo the response
+* **AggregatingDemo** uses a JMS Topic; and aggregates the responses from two inbound gateways, which
+invoke a flow that upper-cases the response; the aggregation returns a list of responses.
 
 
-There are also test cases that exercise both demos; utilizing Spring 3.0 profiles to route the output to a QueueChannel instead of stdout.
+There are also test cases that exercise each demo; utilizing Spring 3.0 profiles to route the output to a QueueChannel instead of stdout.

basic/jms/src/main/java/org/springframework/integration/samples/jms/Main.java

@@ -31,6 +31,7 @@ import org.springframework.context.support.ClassPathXmlApplicationContext;
  *
  * @author Mark Fisher
  * @author Gunnar Hillert
+ * @author Gary Russell
  */
 public class Main {
 
@@ -46,6 +47,11 @@ public class Main {
 		"/META-INF/spring/integration/outboundChannelAdapter.xml"
 	};
 
+	private final static String[] configFilesAggregationDemo = {
+		"/META-INF/spring/integration/common.xml",
+		"/META-INF/spring/integration/aggregation.xml"
+	};
+
 	public static void main(String[] args) {
 
 		final Scanner scanner = new Scanner(System.in);
@@ -64,6 +70,7 @@ public class Main {
 		System.out.println("\n    Which Demo would you like to run? <enter>:\n");
 		System.out.println("\t1. Channel Adapter Demo");
 		System.out.println("\t2. Gateway Demo");
+		System.out.println("\t3. Aggregation Demo");
 
 		while (true) {
 			final String input = scanner.nextLine();
@@ -72,11 +79,18 @@ public class Main {
 				System.out.println("    Loading Channel Adapter Demo...");
 				new ClassPathXmlApplicationContext(configFilesChannelAdapterDemo, Main.class);
 				break;
-			} else if("2".equals(input.trim())) {
+			}
+			else if("2".equals(input.trim())) {
 				System.out.println("    Loading Gateway Demo...");
 				new ClassPathXmlApplicationContext(configFilesGatewayDemo, Main.class);
 				break;
-			} else {
+			}
+			else if("3".equals(input.trim())) {
+				System.out.println("    Loading Aggregation Demo...");
+				new ClassPathXmlApplicationContext(configFilesAggregationDemo, Main.class);
+				break;
+			}
+			else {
 				System.out.println("Invalid choice\n\n");
 				System.out.print("Enter you choice: ");
 			}

basic/jms/src/main/resources/META-INF/spring/integration/aggregation.xml

@@ -0,0 +1,70 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns:int="http://www.springframework.org/schema/integration"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns="http://www.springframework.org/schema/beans"
+	xmlns:int-jms="http://www.springframework.org/schema/integration/jms"
+	xmlns:int-stream="http://www.springframework.org/schema/integration/stream"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans
+			http://www.springframework.org/schema/beans/spring-beans.xsd
+			http://www.springframework.org/schema/context
+			http://www.springframework.org/schema/context/spring-context.xsd
+			http://www.springframework.org/schema/integration
+			http://www.springframework.org/schema/integration/spring-integration.xsd
+			http://www.springframework.org/schema/integration/jms
+			http://www.springframework.org/schema/integration/jms/spring-integration-jms.xsd
+			http://www.springframework.org/schema/integration/stream
+			http://www.springframework.org/schema/integration/stream/spring-integration-stream.xsd">
+
+	<int-stream:stdin-channel-adapter id="stdin" channel="stdinToJmsoutChannel"/>
+
+	<int:channel id="stdinToJmsoutChannel"/>
+
+	<int:chain input-channel="stdinToJmsoutChannel">
+		<int:header-enricher>
+			<int:header name="jms_replyTo" ref="replyQueue" />
+		</int:header-enricher>
+		<int-jms:outbound-channel-adapter destination="requestTopic" />
+	</int:chain>
+
+	<int-jms:message-driven-channel-adapter channel="jmsReplyChannel"
+		destination="replyQueue"/>
+
+	<int:channel id="jmsReplyChannel" />
+
+	<int:aggregator input-channel="jmsReplyChannel" output-channel="out"
+		group-timeout="5000"
+		expire-groups-upon-timeout="false"
+		send-partial-result-on-expiry="true"
+		discard-channel="logLateArrivers"
+		correlation-strategy-expression="headers['jms_correlationId']"
+		release-strategy-expression="size() == 2"/>
+
+	<int:logging-channel-adapter id="logLateArrivers" />
+
+	<!-- Subscribers -->
+
+	<int-jms:inbound-gateway request-channel="upcase" request-destination="requestTopic" />
+
+	<int-jms:inbound-gateway request-channel="upcase" request-destination="requestTopic" />
+
+	<int:transformer input-channel="upcase" expression="payload.toUpperCase()" />
+
+	<!-- Profiles -->
+
+	<beans profile="default">
+
+		<int-stream:stdout-channel-adapter id="out" append-newline="true"/>
+
+	</beans>
+
+	<beans profile="testCase">
+
+		<int:bridge input-channel="out" output-channel="queueChannel"/>
+
+		<int:channel id="queueChannel">
+			<int:queue />
+		</int:channel>
+
+	</beans>
+
+</beans>

basic/jms/src/main/resources/META-INF/spring/integration/common.xml

@@ -14,13 +14,16 @@
 			</bean>
 		</property>
 		<property name="sessionCacheSize" value="10"/>
-		<property name="cacheProducers" value="false"/>
 	</bean>
 
 	<bean id="requestQueue" class="org.apache.activemq.command.ActiveMQQueue">
 		<constructor-arg value="queue.demo"/>
 	</bean>
 
+	<bean id="requestTopic" class="org.apache.activemq.command.ActiveMQTopic">
+		<constructor-arg value="topic.demo"/>
+	</bean>
+
 	<bean id="replyQueue" class="org.apache.activemq.command.ActiveMQQueue">
 		<constructor-arg value="queue.reply"/>
 	</bean>

basic/jms/src/test/java/org/springframework/integration/samples/jms/AggregatorDemoTest.java

@@ -0,0 +1,63 @@
+/*
+ * Copyright 2002-2015 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.samples.jms;
+
+import java.util.List;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import org.springframework.context.support.GenericXmlApplicationContext;
+import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+
+/**
+ * @author Gunnar Hillert
+ * @author Gary Russell
+ */
+public class AggregatorDemoTest {
+
+	private final static String[] configFilesGatewayDemo = {
+		"/META-INF/spring/integration/common.xml",
+		"/META-INF/spring/integration/aggregation.xml"
+	};
+
+	@Test
+	public void testGatewayDemo() throws InterruptedException {
+
+		System.setProperty("spring.profiles.active", "testCase");
+
+		final GenericXmlApplicationContext applicationContext = new GenericXmlApplicationContext(configFilesGatewayDemo);
+
+		final MessageChannel stdinToJmsoutChannel = applicationContext.getBean("stdinToJmsoutChannel", MessageChannel.class);
+
+		stdinToJmsoutChannel.send(MessageBuilder.withPayload("jms test").build());
+
+		final QueueChannel queueChannel = applicationContext.getBean("queueChannel", QueueChannel.class);
+
+		@SuppressWarnings("unchecked")
+		Message<List<String>> reply = (Message<List<String>>) queueChannel.receive(20000);
+		Assert.assertNotNull(reply);
+		List<String> out = reply.getPayload();
+
+		Assert.assertEquals("[JMS TEST, JMS TEST]", out.toString());
+
+		applicationContext.close();
+	}
+
+}