INT-1279 TCP ib and ob Gateways Using Connection Factories - namespace and docs to follow

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpInboundGateway.java

@@ -0,0 +1,89 @@
+/*
+ * Copyright 2002-2010 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.ip.tcp;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.springframework.integration.Message;
+import org.springframework.integration.gateway.AbstractMessagingGateway;
+import org.springframework.integration.ip.IpHeaders;
+import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory;
+import org.springframework.integration.ip.tcp.connection.TcpConnection;
+import org.springframework.integration.ip.tcp.connection.TcpListener;
+import org.springframework.integration.ip.tcp.connection.TcpSender;
+
+/**
+ * Inbound Gateway using a server connection factory - threading is controlled by the
+ * factory. For java.net connections, each socket can process only one message at a time.
+ * For java.nio connections, messages may be multiplexed but the client will need to 
+ * provide correlation logic. If the client is a {@link TcpOutboundGateway} multiplexing
+ * is not used, but multiple concurrent connections can be used if the connection factory uses
+ * single-use connections. For true asynchronous bi-directional communication, a pair of
+ * inbound / outbound channel adapters should be used.
+ * @author Gary Russell
+ * @since 2.0
+ *
+ */
+public class TcpInboundGateway extends AbstractMessagingGateway implements TcpListener, TcpSender {
+
+	protected AbstractServerConnectionFactory connectionFactory;
+	
+	private Map<String, TcpConnection> connections = new ConcurrentHashMap<String, TcpConnection>();
+
+	public boolean onMessage(Message<?> message) {
+		Message<?> reply = this.sendAndReceiveMessage(message);
+		String connectionId = (String) message.getHeaders().get(IpHeaders.CONNECTION_ID);
+		TcpConnection connection = connections.get(connectionId);
+		if (connection == null) {
+			logger.error("Connection " + connectionId + " not found");
+			return false;
+		}
+		try {
+			connection.send(reply);
+		} catch (Exception e) {
+			logger.error("Failed to send reply", e);
+		}
+		return false;
+	}
+
+	/** 
+	 * @return true if the associated connection factory is listening.
+	 */
+	public boolean isListening() {
+		return connectionFactory.isListening();
+	
+	}
+
+	/**
+	 * 
+	 * @param connectionFactory the Connection Factory
+	 */
+	public void setConnectionFactory(AbstractServerConnectionFactory connectionFactory) {
+		this.connectionFactory = connectionFactory;
+		connectionFactory.registerListener(this);
+		connectionFactory.registerSender(this);
+	}
+
+	public void addNewConnection(TcpConnection connection) {
+		connections.put(connection.getConnectionId(), connection);
+	}
+
+	public void removeDeadConnection(TcpConnection connection) {
+		connections.remove(connection.getConnectionId());
+	}
+
+}

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java

@@ -0,0 +1,183 @@
+/*
+ * Copyright 2002-2010 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.ip.tcp;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+
+import org.springframework.integration.Message;
+import org.springframework.integration.MessageTimeoutException;
+import org.springframework.integration.MessagingException;
+import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
+import org.springframework.integration.ip.IpHeaders;
+import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory;
+import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory;
+import org.springframework.integration.ip.tcp.connection.TcpConnection;
+import org.springframework.integration.ip.tcp.connection.TcpListener;
+import org.springframework.integration.ip.tcp.connection.TcpSender;
+import org.springframework.util.Assert;
+
+
+/** 
+ * TCP outbound gateway that uses a client connection factory. If the factory is configured
+ * for single-use connections, each request is sent on a new connection; if the factory does not use
+ * single use connections, each request is blocked until the previous response is received
+ * (or times out). Asynchronous requests/responses over the same connection are not 
+ * supported - use a pair of outbound/inbound adapters for that use case.
+ * 
+ * @author Gary Russell
+ * @since 2.0
+ *
+ */
+public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler
+		implements TcpSender, TcpListener {
+
+	protected AbstractConnectionFactory connectionFactory;
+	
+	private Map<String, AsyncReply> pendingReplies = new ConcurrentHashMap<String, AsyncReply>();
+	
+	private Semaphore semaphore = new Semaphore(1, true);
+
+	private long replyTimeout = 10000;
+	
+	private long requestTimeout = 10000;
+	
+	@Override
+	protected Object handleRequestMessage(Message<?> requestMessage) {
+		Assert.notNull(connectionFactory, this.getClass().getName() +
+				" requires a client connection factory");
+		boolean haveSemaphore = false;
+		try {
+			boolean singleUseConnection = this.connectionFactory.isSingleUse();
+			if (!singleUseConnection) {
+				logger.debug("trying semaphore");
+				if (!this.semaphore.tryAcquire(this.requestTimeout, TimeUnit.MILLISECONDS)) {
+					throw new MessageTimeoutException(requestMessage, "Timed out waiting for connection");
+				}
+				haveSemaphore = true;
+				logger.debug("got semaphore");
+			}
+			TcpConnection connection = this.connectionFactory.getConnection();
+			AsyncReply reply = new AsyncReply();
+			pendingReplies.put(connection.getConnectionId(), reply);
+			if (logger.isDebugEnabled()) {
+				logger.debug("Added " + connection.getConnectionId());
+			}
+			connection.send(requestMessage);
+			Message<?> replyMessage = reply.getReply();
+			if (replyMessage == null) {
+				throw new MessageTimeoutException(requestMessage, "Timed out waiting for response");				
+			}
+			if (logger.isDebugEnabled()) {
+				logger.debug("Respose " + replyMessage);
+			}
+			return replyMessage;
+		} catch (Exception e) {
+			if (e instanceof MessagingException) {
+				throw (MessagingException) e;
+			}
+			throw new MessagingException("Failed to send or receive", e);
+		} finally {
+			if (haveSemaphore) {
+				this.semaphore.release();
+				logger.debug("released semaphore");
+			}
+		}
+	}
+
+	public boolean onMessage(Message<?> message) {
+		String connectionId = (String) message.getHeaders().get(IpHeaders.CONNECTION_ID);
+		if (connectionId == null) {
+			logger.error("Cannot correlate response - no connection id");
+			return false;
+		}
+		AsyncReply reply = pendingReplies.get(connectionId);
+		if (reply == null) {
+			logger.error("Cannot correlate response - no pending reply");
+			return false;
+		}
+		reply.setReply(message);
+		return false;
+	}
+
+	public boolean isListening() {
+		return false;
+	}
+
+	public void setConnectionFactory(AbstractConnectionFactory connectionFactory) {
+		Assert.isTrue(connectionFactory instanceof AbstractClientConnectionFactory, 
+				this.getClass().getName() + " requires a client connection factory");
+		this.connectionFactory = connectionFactory;
+		connectionFactory.registerListener(this);
+		connectionFactory.registerSender(this);
+	}
+
+	public void addNewConnection(TcpConnection connection) {
+		// do nothing - no asynchronous multiplexing supported
+	}
+
+	public void removeDeadConnection(TcpConnection connection) {
+		// do nothing - no asynchronous multiplexing supported
+	}
+
+	/**
+	 * @param replyTimeout the replyTimeout to set
+	 */
+	public void setReplyTimeout(long timeout) {
+		this.replyTimeout = timeout;
+	}
+
+	/**
+	 * @param requestTimeout the requestTimeout to set
+	 */
+	public void setRequestTimeout(long requestTimeout) {
+		this.requestTimeout = requestTimeout;
+	}
+
+	/**
+	 * Class used to coordinate the asynchronous reply to its request.
+	 * @author Gary Russell
+	 * @since 2.0
+	 *
+	 */
+	class AsyncReply {
+		private CountDownLatch latch;
+		private Message<?> reply;
+		public AsyncReply() {
+			this.latch = new CountDownLatch(1);
+		}
+		/**
+		 * Sender blocks here until the reply is received, or we time out
+		 * @return The return message or null if we time out
+		 * @throws Exception
+		 */
+		public Message<?> getReply() throws Exception {
+			if (!this.latch.await(replyTimeout, TimeUnit.MILLISECONDS)) {
+				return null;
+			}
+			return this.reply;
+		}
+		public void setReply(Message<?> reply) {
+			this.reply = reply;
+			latch.countDown();
+		}
+	}
+
+}

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapter.java

@@ -46,8 +46,9 @@ public class TcpReceivingChannelAdapter
 	
 	protected ConnectionFactory serverConnectionFactory;
 	
-	public void onMessage(Message<?> message) {
+	public boolean onMessage(Message<?> message) {
 		sendMessage(message);
+		return false;
 	}
 	
 	@Override

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java

@@ -329,10 +329,13 @@ public abstract class AbstractConnectionFactory
 				this.interceptorFactoryChain.getInterceptorFactories()) {
 			TcpConnectionInterceptor wrapper = factory.getInterceptor();
 			wrapper.setTheConnection(connection);
-			// if no ultimate listener, register each wrapper in turn
+			// if no ultimate listener or sender, register each wrapper in turn
 			if (this.listener == null) {
 				connection.registerListener(wrapper);
 			}
+			if (this.sender == null) {
+				connection.registerSender(wrapper);
+			}
 			connection = wrapper;
 		}
 		return connection;

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnectionInterceptor.java

@@ -110,11 +110,11 @@ public abstract class AbstractTcpConnectionInterceptor implements TcpConnectionI
 		return this.theConnection.isServer();
 	}
 
-	public void onMessage(Message<?> message) {
+	public boolean onMessage(Message<?> message) {
 		if (this.tcpListener == null) {
 			throw new NoListenerException("No listener registered for message reception");
 		}
-		this.tcpListener.onMessage(message);
+		return this.tcpListener.onMessage(message);
 	}
 
 	public void send(Message<?> message) throws Exception {

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpListener.java

@@ -31,7 +31,8 @@ public interface TcpListener {
 	/**
 	 * Called by a TCPConnection when a new message arrives.
 	 * @param message The message.
+	 * @return true if the message was intercepted
 	 */
-	public abstract void onMessage(Message<?> message);
+	public abstract boolean onMessage(Message<?> message);
 	
 }

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java

@@ -103,6 +103,7 @@ public class TcpNetConnection extends AbstractTcpConnection {
 		Message<?> message = null;
 		boolean okToRun = true;
 		logger.debug("Reading...");
+		boolean intercepted = false;
 		while (okToRun) {
 			try {
 				message = this.mapper.toMessage(this);
@@ -127,7 +128,7 @@ public class TcpNetConnection extends AbstractTcpConnection {
 					logger.warn("Unexpected message - no inbound adapter registered with connection " + message);
 					continue;
 				}
-				listener.onMessage(message);
+				intercepted = listener.onMessage(message);
 			} catch (Exception e) {
 				if (e instanceof NoListenerException) {
 					if (this.singleUse) {
@@ -143,9 +144,10 @@ public class TcpNetConnection extends AbstractTcpConnection {
 			}
 			/*
 			 * For single use sockets, we close after receipt if we are on the client
-			 * side, or the server side has no outbound adapter registered
+			 * side, and the data was not intercepted, 
+			 * or the server side has no outbound adapter registered
 			 */
-			if (this.singleUse && this.server && this.sender == null) {
+			if (this.singleUse && ((!this.server && !intercepted) || (this.server && this.sender == null))) {
 				logger.debug("Closing single use socket after inbound message " + this.connectionId);
 				this.close();
 				okToRun = false;

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java

@@ -191,9 +191,10 @@ public class TcpNioConnection extends AbstractTcpConnection {
 			return;
 		}			
 
+		boolean intercepted = false;
 		try {
 			if (message != null) {
-				listener.onMessage(message);
+				intercepted = listener.onMessage(message);
 			}
 		} catch (Exception e) {
 			if (e instanceof NoListenerException) {
@@ -209,9 +210,10 @@ public class TcpNioConnection extends AbstractTcpConnection {
 		}
 		/*
 		 * For single use sockets, we close after receipt if we are on the client
-		 * side, or the server side has no outbound adapter registered
+		 * side, and the data was not intercepted, 
+		 * or the server side has no outbound adapter registered
 		 */
-		if (this.singleUse && this.server && this.sender == null) {
+		if (this.singleUse && ((!this.server && !intercepted) || (this.server && this.sender == null))) {
 			logger.debug("Closing single use cbannel after inbound message " + this.connectionId);
 			this.close();
 		}

spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpInboundGatewayTests.java

@@ -0,0 +1,168 @@
+package org.springframework.integration.ip.tcp;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.Socket;
+
+import javax.net.SocketFactory;
+
+import org.junit.Test;
+import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.core.ChannelResolver;
+import org.springframework.integration.core.MessageChannel;
+import org.springframework.integration.handler.ServiceActivatingHandler;
+import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory;
+import org.springframework.integration.ip.tcp.connection.TcpNetServerConnectionFactory;
+import org.springframework.integration.ip.tcp.connection.TcpNioServerConnectionFactory;
+import org.springframework.integration.ip.util.SocketUtils;
+
+
+public class TcpInboundGatewayTests {
+
+	@Test
+	public void testNetSingle() throws Exception {
+		final int port = SocketUtils.findAvailableServerSocket();
+		AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(port);
+		scf.setSingleUse(true);
+		TcpInboundGateway gateway = new TcpInboundGateway();
+		gateway.setConnectionFactory(scf);
+		scf.start();
+		int n = 0;
+		while (!scf.isListening()) {
+			Thread.sleep(100);
+			if (n++ > 200) {
+				fail("Failed to listen");
+			}
+		}
+		final QueueChannel channel = new QueueChannel();
+		gateway.setRequestChannel(channel);
+		ServiceActivatingHandler handler = new ServiceActivatingHandler(new Service());
+		handler.setChannelResolver(new ChannelResolver() {
+			public MessageChannel resolveChannelName(String channelName) {
+				return channel;
+			}
+		});
+		Socket socket = SocketFactory.getDefault().createSocket("localhost", port);
+		socket.getOutputStream().write("Test1\r\n".getBytes());
+		socket.getOutputStream().write("Test2\r\n".getBytes());
+		handler.handleMessage(channel.receive());
+		handler.handleMessage(channel.receive());
+		byte[] bytes = new byte[12];
+		readFully(socket.getInputStream(), bytes);
+		assertEquals("Echo:Test1\r\n", new String(bytes));
+		readFully(socket.getInputStream(), bytes);
+		assertEquals("Echo:Test2\r\n", new String(bytes));
+	}
+
+	@Test
+	public void testNetNotSingle() throws Exception {
+		final int port = SocketUtils.findAvailableServerSocket();
+		AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(port);
+		scf.setSingleUse(false);
+		TcpInboundGateway gateway = new TcpInboundGateway();
+		gateway.setConnectionFactory(scf);
+		scf.start();
+		int n = 0;
+		while (!scf.isListening()) {
+			Thread.sleep(100);
+			if (n++ > 200) {
+				fail("Failed to listen");
+			}
+		}
+		final QueueChannel channel = new QueueChannel();
+		gateway.setRequestChannel(channel);
+		ServiceActivatingHandler handler = new ServiceActivatingHandler(new Service());
+		Socket socket = SocketFactory.getDefault().createSocket("localhost", port);
+		socket.getOutputStream().write("Test1\r\n".getBytes());
+		socket.getOutputStream().write("Test2\r\n".getBytes());
+		handler.handleMessage(channel.receive());
+		handler.handleMessage(channel.receive());
+		byte[] bytes = new byte[12];
+		readFully(socket.getInputStream(), bytes);
+		assertEquals("Echo:Test1\r\n", new String(bytes));
+		readFully(socket.getInputStream(), bytes);
+		assertEquals("Echo:Test2\r\n", new String(bytes));
+	}
+
+	@Test
+	public void testNioSingle() throws Exception {
+		final int port = SocketUtils.findAvailableServerSocket();
+		AbstractServerConnectionFactory scf = new TcpNioServerConnectionFactory(port);
+		scf.setSingleUse(true);
+		TcpInboundGateway gateway = new TcpInboundGateway();
+		gateway.setConnectionFactory(scf);
+		scf.start();
+		int n = 0;
+		while (!scf.isListening()) {
+			Thread.sleep(100);
+			if (n++ > 200) {
+				fail("Failed to listen");
+			}
+		}
+		final QueueChannel channel = new QueueChannel();
+		gateway.setRequestChannel(channel);
+		ServiceActivatingHandler handler = new ServiceActivatingHandler(new Service());
+		handler.setChannelResolver(new ChannelResolver() {
+			public MessageChannel resolveChannelName(String channelName) {
+				return channel;
+			}
+		});
+		Socket socket = SocketFactory.getDefault().createSocket("localhost", port);
+		socket.getOutputStream().write("Test1\r\n".getBytes());
+		socket.getOutputStream().write("Test2\r\n".getBytes());
+		handler.handleMessage(channel.receive());
+		handler.handleMessage(channel.receive());
+		byte[] bytes = new byte[12];
+		readFully(socket.getInputStream(), bytes);
+		assertEquals("Echo:Test1\r\n", new String(bytes));
+		readFully(socket.getInputStream(), bytes);
+		assertEquals("Echo:Test2\r\n", new String(bytes));
+	}
+
+	@Test
+	public void testNioNotSingle() throws Exception {
+		final int port = SocketUtils.findAvailableServerSocket();
+		AbstractServerConnectionFactory scf = new TcpNioServerConnectionFactory(port);
+		scf.setSingleUse(false);
+		TcpInboundGateway gateway = new TcpInboundGateway();
+		gateway.setConnectionFactory(scf);
+		scf.start();
+		int n = 0;
+		while (!scf.isListening()) {
+			Thread.sleep(100);
+			if (n++ > 200) {
+				fail("Failed to listen");
+			}
+		}
+		final QueueChannel channel = new QueueChannel();
+		gateway.setRequestChannel(channel);
+		ServiceActivatingHandler handler = new ServiceActivatingHandler(new Service());
+		Socket socket = SocketFactory.getDefault().createSocket("localhost", port);
+		socket.getOutputStream().write("Test1\r\n".getBytes());
+		socket.getOutputStream().write("Test2\r\n".getBytes());
+		handler.handleMessage(channel.receive());
+		handler.handleMessage(channel.receive());
+		byte[] bytes = new byte[12];
+		readFully(socket.getInputStream(), bytes);
+		assertEquals("Echo:Test1\r\n", new String(bytes));
+		readFully(socket.getInputStream(), bytes);
+		assertEquals("Echo:Test2\r\n", new String(bytes));
+	}
+
+	private class Service {
+		@SuppressWarnings("unused")
+		public String serviceMethod(byte[] bytes) {
+			return "Echo:" + new String(bytes);
+		}
+	}
+
+	private void readFully(InputStream is, byte[] buff) throws IOException {
+		for (int i = 0; i < buff.length; i++) {
+			buff[i] = (byte) is.read();
+		}
+	}
+	
+}

spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java

@@ -0,0 +1,240 @@
+/*
+ * Copyright 2002-2010 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.ip.tcp;
+
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import javax.net.ServerSocketFactory;
+
+import org.junit.Test;
+import org.springframework.commons.serializer.java.JavaStreamingConverter;
+import org.springframework.integration.Message;
+import org.springframework.integration.MessageTimeoutException;
+import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.core.MessageBuilder;
+import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory;
+import org.springframework.integration.ip.tcp.connection.TcpNetClientConnectionFactory;
+import org.springframework.integration.ip.util.SocketUtils;
+
+
+/**
+ * @author Gary Russell
+ * @since 2.0
+ *
+ */
+public class TcpOutboundGatewayTests {
+
+	@Test 
+	public void testGoodNetSingle() {
+		final int port = SocketUtils.findAvailableServerSocket();
+		AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port);
+		final CountDownLatch latch = new CountDownLatch(1);
+		final AtomicBoolean done = new AtomicBoolean();
+		Executors.newSingleThreadExecutor().execute(new Runnable() {
+			public void run() {
+				try {
+					ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
+					latch.countDown();
+					int i = 0;
+					while (true) {
+						Socket socket = server.accept();
+						ObjectInputStream ois = new ObjectInputStream(socket.getInputStream());
+						Object in = ois.readObject();
+						ObjectOutputStream oos = new ObjectOutputStream(socket.getOutputStream());
+						oos.writeObject("Reply" + (i++));
+					}
+				} catch (Exception e) {
+					if (!done.get()) {
+						e.printStackTrace();
+					}
+				}
+			}
+		});
+		JavaStreamingConverter converter = new JavaStreamingConverter();
+		ccf.setInputConverter(converter);
+		ccf.setOutputConverter(converter);
+		ccf.setSoTimeout(10000);
+		ccf.setSingleUse(true);
+		ccf.setPoolSize(10);
+		ccf.start();
+		TcpOutboundGateway gateway = new TcpOutboundGateway();
+		gateway.setConnectionFactory(ccf);
+		QueueChannel replyChannel = new QueueChannel();
+		gateway.setRequiresReply(true);
+		gateway.setOutputChannel(replyChannel);
+		gateway.setReplyTimeout(60000);
+		gateway.setRequestTimeout(60000);
+		for (int i = 100; i < 200; i++) {
+			gateway.handleMessage(MessageBuilder.withPayload("Test" + i).build());
+		}
+		Set<String> replies = new HashSet<String>();
+		for (int i = 100; i < 200; i++) { 
+			Message<?> m = replyChannel.receive(10000);
+			assertNotNull(m);
+			replies.add((String) m.getPayload());
+		}
+		for (int i = 0; i < 100; i++) {
+			assertTrue(replies.remove("Reply" + i));
+		}
+	}
+
+	@Test 
+	public void testGoodNetMultiplex() {
+		final int port = SocketUtils.findAvailableServerSocket();
+		final CountDownLatch latch = new CountDownLatch(1);
+		final AtomicBoolean done = new AtomicBoolean();
+		Executors.newSingleThreadExecutor().execute(new Runnable() {
+			public void run() {
+				try {
+					ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
+					latch.countDown();
+					int i = 0;
+					Socket socket = server.accept();
+					while (true) {
+						ObjectInputStream ois = new ObjectInputStream(socket.getInputStream());
+						ois.readObject();
+						ObjectOutputStream oos = new ObjectOutputStream(socket.getOutputStream());
+						oos.writeObject("Reply" + (i++));
+					}
+				} catch (Exception e) {
+					if (!done.get()) {
+						e.printStackTrace();
+					}
+				}
+			}
+		});
+		AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port);
+		JavaStreamingConverter converter = new JavaStreamingConverter();
+		ccf.setInputConverter(converter);
+		ccf.setOutputConverter(converter);
+		ccf.setSoTimeout(10000);
+		ccf.setSingleUse(false);
+		ccf.start();
+		TcpOutboundGateway gateway = new TcpOutboundGateway();
+		gateway.setConnectionFactory(ccf);
+		QueueChannel replyChannel = new QueueChannel();
+		gateway.setRequiresReply(true);
+		gateway.setOutputChannel(replyChannel);
+		for (int i = 100; i < 110; i++) {
+			gateway.handleMessage(MessageBuilder.withPayload("Test" + i).build());
+		}
+		Set<String> replies = new HashSet<String>();
+		for (int i = 100; i < 110; i++) { 
+			Message<?> m = replyChannel.receive(10000);
+			assertNotNull(m);
+			replies.add((String) m.getPayload());
+		}
+		for (int i = 0; i < 10; i++) {
+			assertTrue(replies.remove("Reply" + i));
+		}
+	}
+
+	@Test 
+	public void testGoodNetTimeout() throws Exception {
+		final int port = SocketUtils.findAvailableServerSocket();
+		final CountDownLatch latch = new CountDownLatch(1);
+		final AtomicBoolean done = new AtomicBoolean();
+		Executors.newSingleThreadExecutor().execute(new Runnable() {
+			public void run() {
+				try {
+					ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
+					latch.countDown();
+					int i = 0;
+					Socket socket = server.accept();
+					while (true) {
+						ObjectInputStream ois = new ObjectInputStream(socket.getInputStream());
+						ois.readObject();
+						ObjectOutputStream oos = new ObjectOutputStream(socket.getOutputStream());
+						Thread.sleep(1000);
+						oos.writeObject("Reply" + (i++));
+					}
+				} catch (Exception e) {
+					if (!done.get()) {
+						e.printStackTrace();
+					}
+				}
+			}
+		});
+		AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port);
+		JavaStreamingConverter converter = new JavaStreamingConverter();
+		ccf.setInputConverter(converter);
+		ccf.setOutputConverter(converter);
+		ccf.setSoTimeout(10000);
+		ccf.setSingleUse(false);
+		ccf.start();
+		final TcpOutboundGateway gateway = new TcpOutboundGateway();
+		gateway.setConnectionFactory(ccf);
+		gateway.setRequestTimeout(1);
+		QueueChannel replyChannel = new QueueChannel();
+		gateway.setRequiresReply(true);
+		gateway.setOutputChannel(replyChannel);
+		List<Future<Integer>> results = new ArrayList<Future<Integer>>();
+		for (int i = 0; i < 2; i++) {
+			final int j = i;
+			results.add(Executors.newSingleThreadExecutor().submit(new Callable<Integer>(){
+				public Integer call() throws Exception {
+					gateway.handleMessage(MessageBuilder.withPayload("Test" + j).build());
+					return 0;
+				}
+			}));
+		}
+		Set<String> replies = new HashSet<String>();
+		for (int i = 0; i < 2; i++) { 
+			try {
+				results.get(i).get();
+			} catch (InterruptedException e) {
+			} catch (ExecutionException e) {
+				if (i == 0) {
+					fail("Unexpected " + e.getMessage());
+				} else if (i == 1) {
+					assertNotNull(e.getCause());
+					assertTrue(e.getCause() instanceof MessageTimeoutException);
+				}
+				
+				continue;
+			}
+			if (i == 1) {
+				fail("Expected ExecutionException");
+			}
+			Message<?> m = replyChannel.receive(10000);
+			assertNotNull(m);
+			replies.add((String) m.getPayload());
+		}
+		for (int i = 0; i < 1; i++) {
+			assertTrue(replies.remove("Reply" + i));
+		}
+	}
+
+
+}

spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java

@@ -25,8 +25,6 @@ import java.io.InputStream;
 import java.io.ObjectInputStream;
 import java.io.ObjectOutputStream;
 import java.net.Socket;
-import java.net.SocketException;
-import java.net.UnknownHostException;
 import java.util.HashSet;
 import java.util.LinkedList;
 import java.util.List;

spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/HelloWorldInterceptor.java

@@ -58,7 +58,7 @@ public class HelloWorldInterceptor extends AbstractTcpConnectionInterceptor {
 	}
 
 	@Override
-	public void onMessage(Message<?> message) {
+	public boolean onMessage(Message<?> message) {
 		if (!this.negotiated) {
 			Object payload = message.getPayload();
 			if (this.isServer()) {
@@ -67,7 +67,7 @@ public class HelloWorldInterceptor extends AbstractTcpConnectionInterceptor {
 						logger.debug("sending " + this.world);
 						super.send(MessageBuilder.withPayload(world).build());
 						this.negotiated = true;
-						return;
+						return true;
 					} catch (Exception e) {
 						throw new MessagingException("Negotiation error", e);
 					}
@@ -84,10 +84,10 @@ public class HelloWorldInterceptor extends AbstractTcpConnectionInterceptor {
 					throw new MessagingException("Negotiation error - expected '" + world + 
 								"' received " + payload);
 				}
-				return;
+				return true;
 			}
 		}
-		super.onMessage(message);
+		return super.onMessage(message);
 	}
 
 	@Override

spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/MultiClientTests.java

@@ -65,7 +65,7 @@ public class MultiClientTests {
 					UnicastSendingMessageHandler sender = new UnicastSendingMessageHandler(
 							"localhost", adapter.getPort());
 					while (true) {
-						Message message = queueIn.receive();
+						Message<?> message = queueIn.receive();
 						sender.handleMessage(message);
 					}
 				}});
@@ -110,7 +110,7 @@ public class MultiClientTests {
 							SocketUtils.findAvailableUdpSocket(adapter.getPort() + j + 1000),
 							10000);
 					while (true) {
-						Message message = queueIn.receive();
+						Message<?> message = queueIn.receive();
 						sender.handleMessage(message);
 					}
 				}});
@@ -155,7 +155,7 @@ public class MultiClientTests {
 							SocketUtils.findAvailableUdpSocket(adapter.getPort() + j + 1100),
 							10000);
 					while (true) {
-						Message message = queueIn.receive();
+						Message<?> message = queueIn.receive();
 						sender.handleMessage(message);
 					}
 				}});