From e72b101c524de48cdd365f2c6c1afbd8075360fa Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Thu, 17 Jan 2013 10:37:51 -0500 Subject: [PATCH] INT-2886 Fix TCP Connection Exception When an exception occurred while obtaining a connection, the root cause was lost. In addition, a MessageMappingException was thrown instead of a MessageHandlingException. - Deprecate getConnection() in favor of obtainConnection() - Capture the underlying exception as the cause of a MessageHandlingException - Add test case fixed @link (when merging) removed @return with no args (when merging) --- .../ip/tcp/TcpSendingMessageHandler.java | 69 ++++++++++++------- .../ip/tcp/TcpSendingMessageHandlerTests.java | 30 ++++++++ 2 files changed, 76 insertions(+), 23 deletions(-) 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 3420c6ee15..f7a66959cd 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 @@ -32,7 +32,6 @@ import org.springframework.integration.ip.tcp.connection.ClientModeConnectionMan import org.springframework.integration.ip.tcp.connection.ConnectionFactory; import org.springframework.integration.ip.tcp.connection.TcpConnection; import org.springframework.integration.ip.tcp.connection.TcpSender; -import org.springframework.integration.mapping.MessageMappingException; import org.springframework.scheduling.TaskScheduler; import org.springframework.util.Assert; @@ -70,6 +69,11 @@ public class TcpSendingMessageHandler extends AbstractMessageHandler implements private volatile boolean active; + /** + * @deprecated Use {@link #obtainConnection(Message)}. + * TODO: remove in 3.0 + */ + @Deprecated protected TcpConnection getConnection() { TcpConnection connection = null; if (this.clientConnectionFactory == null) { @@ -77,8 +81,22 @@ public class TcpSendingMessageHandler extends AbstractMessageHandler implements } try { connection = this.clientConnectionFactory.getConnection(); - } catch (Exception e) { - logger.error("Error creating SocketWriter", e); + } + catch (Exception e) { + logger.error("Error creating connection", e); + } + return connection; + } + + protected TcpConnection obtainConnection(Message message) { + TcpConnection connection = null; + Assert.notNull(this.clientConnectionFactory, "'clientConnectionFactory' cannot be null"); + try { + connection = this.clientConnectionFactory.getConnection(); + } + catch (Exception e) { + logger.error("Error creating connection", e); + throw new MessageHandlingException(message, "Failed to obtain a connection", e); } return connection; } @@ -101,7 +119,8 @@ public class TcpSendingMessageHandler extends AbstractMessageHandler implements if (connection != null) { try { connection.send(message); - } catch (Exception e) { + } + catch (Exception e) { logger.error("Error sending message", e); connection.close(); if (e instanceof MessageHandlingException) { @@ -111,23 +130,29 @@ public class TcpSendingMessageHandler extends AbstractMessageHandler implements throw new MessageHandlingException(message, "Error sending message", e); } } - } else { + } + else { logger.error("Unable to find outbound socket for " + message); throw new MessageHandlingException(message, "Unable to find outbound socket"); } return; } - - // we own the connection - try { - doWrite(message); - } catch (MessageMappingException e) { - // retry - socket may have closed - if (e.getCause() instanceof IOException) { - logger.debug("Fail on first write attempt", e); + else { + // we own the connection + try { doWrite(message); - } else { - throw e; + } + catch (MessageHandlingException e) { + // retry - socket may have closed + if (e.getCause() instanceof IOException) { + if (logger.isDebugEnabled()) { + logger.debug("Fail on first write attempt", e); + } + doWrite(message); + } + else { + throw e; + } } } } @@ -139,23 +164,21 @@ public class TcpSendingMessageHandler extends AbstractMessageHandler implements protected void doWrite(Message message) { TcpConnection connection = null; try { - connection = getConnection(); - if (connection == null) { - throw new MessageMappingException(message, "Failed to create connection"); - } + connection = obtainConnection(message); if (logger.isDebugEnabled()) { logger.debug("Got Connection " + connection.getConnectionId()); } connection.send(message); - } catch (Exception e) { + } + catch (Exception e) { String connectionId = null; if (connection != null) { connectionId = connection.getConnectionId(); } - if (e instanceof MessageMappingException) { - throw (MessageMappingException) e; + if (e instanceof MessageHandlingException) { + throw (MessageHandlingException) e; } - throw new MessageMappingException(message, "Failed to map message using " + connectionId, e); + throw new MessageHandlingException(message, "Failed to handle message using " + connectionId, e); } } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java index db19e88043..ae853eec5f 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java @@ -20,6 +20,7 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; +import static org.mockito.Mockito.mock; import java.io.IOException; import java.io.InputStream; @@ -27,6 +28,7 @@ import java.io.ObjectInputStream; import java.io.ObjectOutputStream; import java.net.ServerSocket; import java.net.Socket; +import java.net.SocketException; import java.util.ArrayList; import java.util.HashSet; import java.util.List; @@ -43,14 +45,19 @@ import javax.net.ServerSocketFactory; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.junit.Test; +import org.mockito.Mockito; +import org.mockito.invocation.InvocationOnMock; +import org.mockito.stubbing.Answer; import org.springframework.context.ApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.core.serializer.DefaultDeserializer; import org.springframework.core.serializer.DefaultSerializer; import org.springframework.integration.Message; import org.springframework.integration.MessageChannel; +import org.springframework.integration.MessagingException; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.core.PollableChannel; +import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory; import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory; import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory; import org.springframework.integration.ip.tcp.connection.HelloWorldInterceptorFactory; @@ -67,6 +74,7 @@ import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.test.util.SocketUtils; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; + /** * @author Gary Russell * @author Artem Bilan @@ -1095,4 +1103,26 @@ public class TcpSendingMessageHandlerTests { assertEquals(testPayload, new String((byte[]) m.getPayload())); } + @Test + public void testConnectionException() throws Exception { + TcpSendingMessageHandler handler = new TcpSendingMessageHandler(); + AbstractConnectionFactory mockCcf = mock(AbstractClientConnectionFactory.class); + Mockito.doAnswer(new Answer() { + + public Object answer(InvocationOnMock invocation) throws Throwable { + throw new SocketException("Failed to connect"); + } + }).when(mockCcf).getConnection(); + handler.setConnectionFactory(mockCcf); + try { + handler.handleMessage(new GenericMessage("foo")); + fail("Expected exception"); + } + catch (Exception e) { + assertTrue(e instanceof MessagingException); + assertTrue(e.getCause() != null); + assertTrue(e.getCause() instanceof SocketException); + assertEquals("Failed to connect", e.getCause().getMessage()); + } + } }