From 1cb7990f6c32536098de041c60ba232afea84ab6 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Thu, 13 Oct 2011 16:54:32 -0400 Subject: [PATCH] INT-2179 Fix NPE If Connection Id Header Missing ip_connection_id header is used to determine to which socket a message should be written. If this header is missing, we can't send the message and we log an error. Previously, an NPE was thrown. This applies to inbound gateway replies and asynchronous outbound server messages. --- docs/src/reference/docbook/ip.xml | 40 ++++++++++++++++--- .../integration/ip/tcp/TcpInboundGateway.java | 7 +++- .../ip/tcp/TcpSendingMessageHandler.java | 7 +++- 3 files changed, 45 insertions(+), 9 deletions(-) diff --git a/docs/src/reference/docbook/ip.xml b/docs/src/reference/docbook/ip.xml index b7f357fef9..369bbcd980 100644 --- a/docs/src/reference/docbook/ip.xml +++ b/docs/src/reference/docbook/ip.xml @@ -486,7 +486,19 @@ from the response message by writing it to the connection. - The outbound gateway, after sending a message over the connection, waits for a response and + + + For the inbound gateway, care must be taken to retain, or populate, the + ip_connnection_id header because it is used to + correlate the message to a connection. Messages that originate at the + gateway will automatically have the header set. If the reply is + constructed as a new message, you will need to set the header. The + header value can be captured from the incoming message. + + + + + The outbound gateway, after sending a message over the connection, waits for a response and constructs a response message and puts in on the reply channel. Communications over the connections are single-threaded. Users should be aware that only one message can be handled at a time and, if another thread attempts to send @@ -562,12 +574,30 @@ To achieve high-volume throughput (avoiding the pitfalls of using gateways as mentioned above) you may consider configuring a pair of collaborating - outbound and inbound channel adapters. On the server side, message + outbound and inbound channel adapters. + Collaborating adapters can also be used (server-side or client-side) for + totally asynchronous communication (rather than with request/reply semantics). + On the server side, message correlation is automatically handled by the adapters because the inbound adapter adds a header allowing the outbound adapter to determine which - connection to use to send the reply message. On the client side, however, - the application will have to provide its own correlation logic. This can - be done in a number of ways. + connection to use to send the reply message. + + + + + On the server side, care must be taken to populate the + ip_connnection_id header because it is used to + correlate the message to a connection. Messages that originate at the + inbound adapter will automatically have the header set. If you wish to + construct other messages to send, you will need to set the header. The + header value can be captured from an incoming message. + + + + + On the client side, + the application will have to provide its own correlation logic, if needed. + This can be done in a number of ways. If the message payload has some natural correlation data, such as a diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpInboundGateway.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpInboundGateway.java index 9c74b8113d..8cf4d50b99 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpInboundGateway.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpInboundGateway.java @@ -53,9 +53,12 @@ public class TcpInboundGateway extends MessagingGatewaySupport implements TcpLis return false; } String connectionId = (String) message.getHeaders().get(IpHeaders.CONNECTION_ID); - TcpConnection connection = connections.get(connectionId); + TcpConnection connection = null; + if (connectionId != null) { + connection = connections.get(connectionId); + } if (connection == null) { - logger.error("Connection " + connectionId + " not found when processing reply for " + message); + logger.error("Connection not found when processing reply " + reply + " for " + message); return false; } try { diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandler.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandler.java index 83dd53b3b7..1370e92ed8 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandler.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandler.java @@ -81,7 +81,10 @@ public class TcpSendingMessageHandler extends AbstractMessageHandler implements if (this.serverConnectionFactory != null) { // We don't own the connection, we are asynchronously replying Object connectionId = message.getHeaders().get(IpHeaders.CONNECTION_ID); - TcpConnection connection = connections.get(connectionId); + TcpConnection connection = null; + if (connectionId != null) { + connection = connections.get(connectionId); + } if (connection != null) { try { connection.send(message); @@ -90,7 +93,7 @@ public class TcpSendingMessageHandler extends AbstractMessageHandler implements connection.close(); } } else { - logger.error("Unable to find incoming socket for " + message); + logger.error("Unable to find outbound socket for " + message); } return; }