Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-1458 Add Sequence Header to Tcp Messages To Facilitate Resequencing

Browse Source
This commit is contained in:
Gary Russell
2010-09-23 21:02:16 -04:00
parent 76fc874be4
commit 3764d30911
8 changed files with 94 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-integration-ip/src/main/java/org/springframework/integration/ip/IpHeaders.java
View File

@@ -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";
}

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

@@ -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();
}
}

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

@@ -156,4 +156,9 @@ public abstract class AbstractTcpConnectionInterceptor implements TcpConnectionI
}
}
public long getConnectionSeq() {
return this.theConnection.getConnectionSeq();
}
}

5
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnection.java
View File

@@ -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();
}

1
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapper.java
View File

@@ -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;

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

@@ -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;

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

@@ -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;

54
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapperTests.java
View File

@@ -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<Object> 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