diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnection.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnection.java
index 142f5c24b2..963af34fe0 100644
--- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnection.java
+++ b/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;
+ }
+
}
diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java
index ee3dc6b8d4..43a3333cbf 100644
--- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java
+++ b/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());
}
diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java
index 9fa9611cd4..c658bcdb12 100644
--- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java
+++ b/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) {
diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests-context.xml b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests-context.xml
index e2f11b9e49..9541538c09 100644
--- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests-context.xml
+++ b/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">
@@ -13,8 +15,11 @@
+
+
diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests.java
index e12e788e8c..d4e3c80012 100644
--- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests.java
+++ b/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
diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionTests.java
index 8a7b85552c..2c6ee11fdf 100644
--- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionTests.java
+++ b/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) {