INT-1797 Polishing

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

@@ -16,6 +16,7 @@
 
 package org.springframework.integration.ip.tcp.connection;
 
+import java.net.InetAddress;
 import java.net.Socket;
 import java.net.SocketException;
 import java.util.concurrent.atomic.AtomicLong;
@@ -64,9 +65,18 @@ public abstract class AbstractTcpConnection implements TcpConnection {
 	private AtomicLong sequence = new AtomicLong();
 	
 	private int soLinger = -1;
+
+	private String hostName = "unknown";
+
+	private String hostAddress = "unknown";
 	
 	public AbstractTcpConnection(Socket socket, boolean server) {
 		this.server = server;
+		InetAddress inetAddress = socket.getInetAddress();
+		if (inetAddress != null) {
+			this.hostAddress = inetAddress.getHostAddress();
+			this.hostName = inetAddress.getHostName();
+		}
 		try {
 			this.soLinger = socket.getSoLinger();
 		} catch (SocketException e) { }
@@ -221,5 +231,13 @@ public abstract class AbstractTcpConnection implements TcpConnection {
 		return sequence.incrementAndGet();
 	}
 
+	public String getHostAddress() {
+		return this.hostAddress;
+	}
+
+	public String getHostName() {
+		return this.hostName;
+	}
+
 	
 }

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

@@ -16,6 +16,7 @@
 
 package org.springframework.integration.ip.tcp.connection;
 
+import java.net.InetAddress;
 import java.net.Socket;
 import java.net.SocketTimeoutException;
 
@@ -71,14 +72,6 @@ public class TcpNetConnection extends AbstractTcpConnection {
 		this.afterSend(message);
 	}
 
-	public String getHostAddress() {
-		return this.socket.getInetAddress().getHostAddress();
-	}
-
-	public String getHostName() {
-		return this.socket.getInetAddress().getHostName();
-	}
-
 	public Object getPayload() throws Exception {
 		return this.deserializer.deserialize(this.socket.getInputStream());
 	}

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

@@ -45,26 +45,26 @@ public class TcpNioConnection extends AbstractTcpConnection {
 
 	private final SocketChannel socketChannel;
 	
-	private OutputStream channelOutputStream;
+	private volatile OutputStream channelOutputStream;
 	
-	private PipedOutputStream pipedOutputStream;
+	private volatile PipedOutputStream pipedOutputStream;
 	
-	private PipedInputStream pipedInputStream;
+	private volatile PipedInputStream pipedInputStream;
 
-	private boolean usingDirectBuffers;
+	private volatile boolean usingDirectBuffers;
 	
-	private Executor taskExecutor;
+	private volatile Executor taskExecutor;
 	
-	private ByteBuffer rawBuffer;
+	private volatile ByteBuffer rawBuffer;
 	
-	private int maxMessageSize = 60 * 1024;
+	private volatile int maxMessageSize = 60 * 1024;
 	
-	private long lastRead;
+	private volatile long lastRead;
 	
 	private AtomicInteger executionControl = new AtomicInteger();
 	
-	private boolean writingToPipe;
-
+	private volatile boolean writingToPipe;
+	
 	/**
 	 * Constructs a TcpNetConnection for the SocketChannel.
 	 * @param socketChannel the socketChannel
@@ -112,14 +112,6 @@ public class TcpNioConnection extends AbstractTcpConnection {
 		}
 	}
 
-	public String getHostAddress() {
-		return this.socketChannel.socket().getInetAddress().getHostAddress();
-	}
-
-	public String getHostName() {
-		return this.socketChannel.socket().getInetAddress().getHostName();
-	}
-
 	public Object getPayload() throws Exception {
 		return this.deserializer.deserialize(pipedInputStream);
 	}
@@ -151,7 +143,9 @@ public class TcpNioConnection extends AbstractTcpConnection {
 	 * sockets.
 	 */
 	public void run() {
-		logger.trace("Nio message assembler running...");
+		if (logger.isTraceEnabled()) {
+			logger.trace(this.getConnectionId() + " Nio message assembler running...");
+		}
 		try {
 			if (this.listener == null && !this.singleUse) {
 				logger.debug("TcpListener exiting - no listener and not single use");			
@@ -193,8 +187,7 @@ public class TcpNioConnection extends AbstractTcpConnection {
 	}
 
 	private boolean dataAvailable() throws IOException {
-		return this.socketChannel.isOpen() && 
-			   (this.pipedInputStream.available() > 0 || writingToPipe);
+		return this.pipedInputStream.available() > 0 || writingToPipe;
 	}
 
 	/**
@@ -264,28 +257,29 @@ public class TcpNioConnection extends AbstractTcpConnection {
 	}
 	
 	private void doRead() throws Exception {
-		if (rawBuffer == null) {
-			rawBuffer = allocate(maxMessageSize);
+		if (this.rawBuffer == null) {
+			this.rawBuffer = allocate(maxMessageSize);
 		}
 
-		writingToPipe = true;
+		this.writingToPipe = true;
 		if (this.taskExecutor == null) {
 			this.taskExecutor = Executors.newSingleThreadExecutor();
 		}
 		// If there is no assembler running, start one
 		checkForAssembler();
-		rawBuffer.clear();
-		int len = socketChannel.read(rawBuffer);
+		this.rawBuffer.clear();
+		int len = this.socketChannel.read(this.rawBuffer);
 		if (len < 0) {
+			this.writingToPipe = false;
 			this.closeConnection();
 		}
-		rawBuffer.flip();
+		this.rawBuffer.flip();
 		if (logger.isDebugEnabled()) {
 			logger.debug("Read " + rawBuffer.limit() + " into raw buffer");
 		}
-		pipedOutputStream.write(rawBuffer.array(), 0, rawBuffer.limit());
-		pipedOutputStream.flush();
-		writingToPipe = false;
+		this.pipedOutputStream.write(this.rawBuffer.array(), 0, this.rawBuffer.limit());
+		this.pipedOutputStream.flush();
+		this.writingToPipe = false;
 		
 	}
 
@@ -305,7 +299,9 @@ public class TcpNioConnection extends AbstractTcpConnection {
 	 * Invoked by the factory when there is data to be read.
 	 */
 	public void readPacket() {
-		logger.debug("Reading...");
+		if (logger.isDebugEnabled()) {
+			logger.debug(this.getConnectionId() + " Reading...");
+		}
 		try {
 			doRead();
 		} catch (ClosedChannelException cce) {

spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests-context.xml

@@ -3,8 +3,10 @@
 	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
 	xmlns:int="http://www.springframework.org/schema/integration"
 	xmlns:int-ip="http://www.springframework.org/schema/integration/ip"
-	xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
-		http://www.springframework.org/schema/integration/ip http://www.springframework.org/schema/integration/ip/spring-integration-ip.xsd
+	xmlns:task="http://www.springframework.org/schema/task"
+	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/task http://www.springframework.org/schema/task/spring-task-3.0.xsd
 		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd">
 
 	<int:message-history/>
@@ -13,8 +15,11 @@
 
 	<int-ip:tcp-connection-factory id="server"
 		type="server"
+		using-nio="true"
 		single-use="true"
 		port="#{tcpIpUtils.findAvailableServerSocket(10000)}"
+		task-executor="exec"
+		so-timeout="20000"
 	/>
 
 	<int-ip:tcp-connection-factory id="client"
@@ -28,10 +33,13 @@
 	<int-ip:tcp-inbound-gateway id="looper"
 		request-channel="queue"
 		connection-factory="server"
+		reply-timeout="1"
 		/>
 			
 	<int:channel id="queue">
 		<int:queue/>
 	</int:channel>
 	
+	<task:executor id="exec" pool-size="10"/>
+	
 </beans>

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

@@ -69,16 +69,18 @@ public class ConnectionToConnectionTests {
 				throw new Exception("Failed to listen");
 			}
 		}
-		TcpConnection connection = client.getConnection();
-		connection.send(MessageBuilder.withPayload("Test").build());
-		Message<?> message = serverSideChannel.receive(10000);
-		MessageHistory history = MessageHistory.read(message);
-		//org.springframework.integration.test.util.TestUtils
-		Properties componentHistoryRecord = TestUtils.locateComponentInHistory(history, "looper", 0);
-		assertNotNull(componentHistoryRecord);
-		assertTrue(componentHistoryRecord.get("type").equals("ip:tcp-inbound-gateway"));
-		assertNotNull(message);
-		assertEquals("Test", new String((byte[]) message.getPayload()));
+		for (int i = 0; i < 100; i++) {
+			TcpConnection connection = client.getConnection();
+			connection.send(MessageBuilder.withPayload("Test").build());
+			Message<?> message = serverSideChannel.receive(10000);
+			MessageHistory history = MessageHistory.read(message);
+			//org.springframework.integration.test.util.TestUtils
+			Properties componentHistoryRecord = TestUtils.locateComponentInHistory(history, "looper", 0);
+			assertNotNull(componentHistoryRecord);
+			assertTrue(componentHistoryRecord.get("type").equals("ip:tcp-inbound-gateway"));
+			assertNotNull(message);
+			assertEquals("Test", new String((byte[]) message.getPayload()));
+		}
 	}
 
 	@Test

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

@@ -54,7 +54,7 @@ public class TcpNioConnectionTests {
 				try {
 					ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
 					latch.countDown();
-					server.accept();
+					Socket s = server.accept();
 					// block so we fill the buffer
 					server.accept();
 				} catch (Exception e) {