From 3764d309110c9a5a201eeb90dea978f312f08d87 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Thu, 23 Sep 2010 21:02:16 -0400 Subject: [PATCH] INT-1458 Add Sequence Header to Tcp Messages To Facilitate Resequencing --- .../integration/ip/IpHeaders.java | 2 + .../tcp/connection/AbstractTcpConnection.java | 9 ++++ .../AbstractTcpConnectionInterceptor.java | 5 ++ .../ip/tcp/connection/TcpConnection.java | 5 ++ .../ip/tcp/connection/TcpMessageMapper.java | 1 + .../ip/tcp/connection/TcpNetConnection.java | 13 +++-- .../ip/tcp/connection/TcpNioConnection.java | 14 +++-- .../tcp/connection/TcpMessageMapperTests.java | 54 +++++++++++++++++++ 8 files changed, 94 insertions(+), 9 deletions(-) diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/IpHeaders.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/IpHeaders.java index 20788b9a64..b316e3d934 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/IpHeaders.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/IpHeaders.java @@ -46,4 +46,6 @@ public abstract class IpHeaders { public static final String CONNECTION_ID = IP + "connection_id"; + public static final String CONNECTION_SEQ = IP + "connection_seq"; + } 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 56fe8c3e0f..45fd6d4f61 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,8 @@ package org.springframework.integration.ip.tcp.connection; +import java.util.concurrent.atomic.AtomicLong; + import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.commons.serializer.InputStreamingConverter; @@ -54,6 +56,8 @@ public abstract class AbstractTcpConnection implements TcpConnection { protected String connectionId; + private AtomicLong sequence = new AtomicLong(); + public AbstractTcpConnection(boolean server) { this.server = server; } @@ -163,4 +167,9 @@ public abstract class AbstractTcpConnection implements TcpConnection { return server; } + public long getConnectionSeq() { + return sequence.incrementAndGet(); + } + + } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnectionInterceptor.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnectionInterceptor.java index 079ae4b111..30e7143f2c 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnectionInterceptor.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnectionInterceptor.java @@ -156,4 +156,9 @@ public abstract class AbstractTcpConnectionInterceptor implements TcpConnectionI } } + public long getConnectionSeq() { + return this.theConnection.getConnectionSeq(); + } + + } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnection.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnection.java index 2c2ae3e7f6..d6e126e33c 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnection.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnection.java @@ -143,5 +143,10 @@ public interface TcpConnection extends Runnable { * @return this connection's listener */ public TcpListener getListener(); + + /** + * @return the next sequence number for a message received on this socket + */ + public long getConnectionSeq(); } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapper.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapper.java index 56bc8726eb..13d2f229ef 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapper.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapper.java @@ -53,6 +53,7 @@ public class TcpMessageMapper implements .setHeader(IpHeaders.IP_ADDRESS, connection.getHostAddress()) .setHeader(IpHeaders.REMOTE_PORT, connection.getPort()) .setHeader(IpHeaders.CONNECTION_ID, connection.getConnectionId()) + .setHeader(IpHeaders.CONNECTION_SEQ, connection.getConnectionSeq()) .build(); } return message; 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 794cb3ef5d..66e1ef38c5 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 @@ -113,10 +113,15 @@ public class TcpNetConnection extends AbstractTcpConnection { if (e instanceof SocketTimeoutException && this.singleUse) { logger.debug("Closing single use socket after timeout"); } else { - logger.error("Read exception " + - this.getConnectionId() + " " + - e.getClass().getSimpleName() + - ":" + e.getCause() + ":" + e.getMessage()); + if (logger.isTraceEnabled()) { + logger.error("Read exception " + + this.getConnectionId(), e); + } else { + logger.error("Read exception " + + this.getConnectionId() + " " + + e.getClass().getSimpleName() + + ":" + e.getCause() + ":" + e.getMessage()); + } } } break; 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 d0f5dac9a1..b751a95e0c 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 @@ -216,11 +216,15 @@ public class TcpNioConnection extends AbstractTcpConnection { } } else { if (!(e instanceof SoftEndOfStreamException)) { - logger.error("Read exception " + - this.getConnectionId() + " " + - e.getClass().getSimpleName() + - ":" + e.getCause() + ":" + e.getMessage()); - + if (logger.isTraceEnabled()) { + logger.error("Read exception " + + this.getConnectionId(), e); + } else { + logger.error("Read exception " + + this.getConnectionId() + " " + + e.getClass().getSimpleName() + + ":" + e.getCause() + ":" + e.getMessage()); + } } } return null; diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapperTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapperTests.java index 6ce072551e..76f2e66fb0 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapperTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapperTests.java @@ -60,6 +60,60 @@ public class TcpMessageMapperTests { .getHeaders().get(IpHeaders.REMOTE_PORT)); } + /** + * Test method for {@link org.springframework.integration.ip.tcp.SocketMessageMapper#toMessage(org.springframework.integration.ip.tcp.SocketReader)}. + * Tests segmented reads into the payload and verifies reassembly. + */ + @Test + public void testToMessageSequence() throws Exception { + + TcpMessageMapper mapper = new TcpMessageMapper(); + TcpConnection connection = new AbstractTcpConnection(false) { + public void run() { + } + public void send(Message message) throws Exception { + } + public boolean isOpen() { + return false; + } + public int getPort() { + return 1234; + } + public Object getPayload() throws Exception { + return TEST_PAYLOAD.getBytes(); + } + public String getHostName() { + return "MyHost"; + } + public String getHostAddress() { + return "1.1.1.1"; + } + public String getConnectionId() { + return "anId"; + } + }; + Message message = mapper.toMessage(connection); + assertEquals(TEST_PAYLOAD, new String((byte[]) message.getPayload())); + assertEquals("MyHost", message + .getHeaders().get(IpHeaders.HOSTNAME)); + assertEquals("1.1.1.1", message + .getHeaders().get(IpHeaders.IP_ADDRESS)); + assertEquals(1234, message + .getHeaders().get(IpHeaders.REMOTE_PORT)); + assertEquals(1L, message + .getHeaders().get(IpHeaders.CONNECTION_SEQ)); + message = mapper.toMessage(connection); + assertEquals(TEST_PAYLOAD, new String((byte[]) message.getPayload())); + assertEquals("MyHost", message + .getHeaders().get(IpHeaders.HOSTNAME)); + assertEquals("1.1.1.1", message + .getHeaders().get(IpHeaders.IP_ADDRESS)); + assertEquals(1234, message + .getHeaders().get(IpHeaders.REMOTE_PORT)); + assertEquals(2L, message + .getHeaders().get(IpHeaders.CONNECTION_SEQ)); + } + /** * Test method for {@link org.springframework.integration.ip.tcp.SocketMessageMapper#fromMessage(org.springframework.integration.Message)}. * @throws Exception