From 69b4a0db7808e9dfa818935d554c9bfd74405692 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Sun, 20 Feb 2011 20:29:44 -0500 Subject: [PATCH] INT-1797 Polishing --- .../tcp/connection/AbstractTcpConnection.java | 18 ++++++ .../ip/tcp/connection/TcpNetConnection.java | 9 +-- .../ip/tcp/connection/TcpNioConnection.java | 58 +++++++++---------- .../ConnectionToConnectionTests-context.xml | 12 +++- .../ip/tcp/ConnectionToConnectionTests.java | 22 +++---- .../tcp/connection/TcpNioConnectionTests.java | 2 +- 6 files changed, 69 insertions(+), 52 deletions(-) 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) {