TCP Async - Add Timeout Test

http://stackoverflow.com/questions/32614395/spring-integration-tcp-outbound-channel-adapter-that-detects-connection-loss-and/32617568#comment54590607_32617568

Polishing - PR Comments

intermediate/tcp-client-server-multiplex/README.md

@@ -1,10 +1,43 @@
 TCP Client-Server Multiplex Sample
 ==================================
 
-If this is your first experience with the spring-integrtion-ip module, start with the **tcp-client-server** project in the basic folder.
+If this is your first experience with the spring-integration-ip module, start with the **tcp-client-server** project in the basic folder.
 
 That project uses outbound and inbound tcp gateways for communication. As discussed in the Spring Integration Reference Manual, this has some limitations for performance. If a shared socket (single-use="false") is used, only one message can be processed at a time (on the client side); we must wait for the response to the current request before we can send the next request. Otherwise, because only the payload is sent over tcp, the framework cannot correlate responses to requests.
 
 An alternative is to use a new socket for each message, but this comes with a performance overhead. The solution is to use **Collaborating Channel Adapters** (see SI Reference Manual). In such a scenario, we can send multiple requests before a response is received. This is termed multiplexing.
 
-This sample demonstrates how to configure collaborating channel adapters, on both the client and server sides, and one technique for correlating the responses to the corresponding request.
+This sample demonstrates how to configure collaborating channel adapters, on both the client and server sides, and one
+technique for correlating the responses to the corresponding request.
+
+````
+gateway -> outbound-channel-adapter
+        |-> aggregator
+
+inbound-channel-adapter->aggregator->transformer
+````
+
+When the aggregator receives the reply, the group is released and transformed to just the reply which is then returned
+to the gateway.
+
+Unlike when using TCP gateways, there is no way to communicate an IO error to the waiting thread, which is sitting in
+the initial `<gateway/>` waiting for a reply - it "knows" nothing about the downstream flow, such as a read timeout
+on the socket.
+
+This sample now shows how to use the `group-timeout` on the aggregator to release the group under this condition.
+Further, it routes the discarded message to a service activator which return a `MessagingTimeoutException` which
+is routed to the waiting thread.
+
+````
+gateway -> outbound-channel-adapter
+        |-> aggregator
+
+aggregator(group-timeout discard)->service-activator
+````
+
+A service activator is used here instead of a transformer because you may wish to take some other action when the
+timeout condition occurs.
+
+Normal gateway processing detects that the payload is an `Exception` and throws it to the caller.
+
+Thus, this shows how to return an exception to a gateway caller, even when the messaging is entirely asynchronous.

intermediate/tcp-client-server-multiplex/src/main/java/org/springframework/integration/samples/tcpclientserver/EchoService.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2010 the original author or authors.
+ * 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.
@@ -15,6 +15,8 @@
  */
 package org.springframework.integration.samples.tcpclientserver;
 
+import org.springframework.integration.MessageTimeoutException;
+
 /**
  * Simple service that receives data in a byte array,
  * converts it to a String and appends it with ':echo'.
@@ -25,11 +27,19 @@ package org.springframework.integration.samples.tcpclientserver;
  */
 public class EchoService {
 
-	public String test(String input) {
+	public String test(String input) throws InterruptedException {
 		if ("FAIL".equals(input)) {
 			throw new RuntimeException("Failure Demonstration");
 		}
+		else if("TIMEOUT_TEST".equals(input)){
+			Thread.sleep(3000);
+		}
+
 		return input + ":echo";
 	}
 
-}
+	public MessageTimeoutException noResponse(String input) {
+		return new MessageTimeoutException("No response received for " + input);
+	}
+
+}

intermediate/tcp-client-server-multiplex/src/main/resources/META-INF/spring/integration/tcpClientServerDemo-conversion-context.xml

@@ -4,10 +4,10 @@
 	xmlns="http://www.springframework.org/schema/integration"
 	xmlns:ip="http://www.springframework.org/schema/integration/ip"
 	xmlns:context="http://www.springframework.org/schema/context"
-	xsi:schemaLocation="http://www.springframework.org/schema/integration/ip http://www.springframework.org/schema/integration/ip/spring-integration-ip.xsd
-		http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
-		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">
+	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/ip http://www.springframework.org/schema/integration/ip/spring-integration-ip.xsd
+		http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd">
 
 	<beans:description>
 		Uses conversion service and collaborating channel adapters.
@@ -29,8 +29,9 @@
 	<!-- Client side -->
 
 	<gateway id="gw"
-        service-interface="org.springframework.integration.samples.tcpclientserver.SimpleGateway"
-        default-request-channel="input" />
+		service-interface="org.springframework.integration.samples.tcpclientserver.SimpleGateway"
+		default-reply-timeout="20000"
+		default-request-channel="input" />
 
 	<ip:tcp-connection-factory id="client"
 		type="client"
@@ -65,9 +66,17 @@
 
 	<aggregator input-channel="toAggregator.client"
 		output-channel="toTransformer.client"
+		expire-groups-upon-completion="true"
+		expire-groups-upon-timeout="true"
+		discard-channel="noResponseChannel"
+		group-timeout="1000"
 		correlation-strategy-expression="payload.substring(0,3)"
 		release-strategy-expression="size() == 2" />
 
+	<channel id="noResponseChannel" />
+
+	<service-activator input-channel="noResponseChannel" ref="echoService" method="noResponse" />
+
 	<transformer input-channel="toTransformer.client"
 		expression="payload.get(1)"/> <!-- The response is always second -->
 

intermediate/tcp-client-server-multiplex/src/test/java/org/springframework/integration/samples/tcpclientserver/TcpClientServerDemoTest.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2010 the original author or authors.
+ * 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.
@@ -15,8 +15,11 @@
  */
 package org.springframework.integration.samples.tcpclientserver;
 
+import static org.hamcrest.Matchers.containsString;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
 
 import java.util.HashSet;
 import java.util.Set;
@@ -26,9 +29,11 @@ import java.util.concurrent.TimeUnit;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
+
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.core.task.SimpleAsyncTaskExecutor;
 import org.springframework.core.task.TaskExecutor;
+import org.springframework.integration.MessageTimeoutException;
 import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory;
 import org.springframework.integration.ip.util.TestingUtilities;
 import org.springframework.integration.samples.tcpclientserver.support.CustomTestContextLoader;
@@ -79,6 +84,7 @@ public class TcpClientServerDemoTest {
 			results.add(i);
 			final int j = i;
 			executor.execute(new Runnable() {
+				@Override
 				public void run() {
 					String result = gw.send(j + "Hello world!"); // first 3 bytes is correlationid
 					assertEquals(j + "Hello world!:echo", result);
@@ -89,4 +95,16 @@ public class TcpClientServerDemoTest {
 		assertTrue(latch.await(10, TimeUnit.SECONDS));
 		assertEquals(0, results.size());
 	}
+
+	@Test
+	public void testTimeout() {
+		try {
+			gw.send("TIMEOUT_TEST");
+			fail("expected exception");
+		}
+		catch (MessageTimeoutException e) {
+			assertThat(e.getMessage(), containsString("No response received for TIMEOUT_TEST"));
+		}
+	}
+
 }