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 bbd71c52a0..f30d66a882 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 @@ -100,6 +100,7 @@ public class TcpNetConnection extends TcpConnectionSupport { } catch (Exception e) { this.publishConnectionExceptionEvent(e); + this.closeConnection(true); throw e; } this.afterSend(message); 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 25c0ff1e34..459b8a15ff 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 @@ -134,7 +134,7 @@ public class TcpNioConnection extends TcpConnectionSupport { @SuppressWarnings("unchecked") public void send(Message message) throws Exception { - synchronized(this.getMapper()) { + synchronized(this.socketChannel) { Object object = this.getMapper().fromMessage(message); this.lastSend = System.currentTimeMillis(); try { @@ -142,6 +142,7 @@ public class TcpNioConnection extends TcpConnectionSupport { } catch (Exception e) { this.publishConnectionExceptionEvent(e); + this.closeConnection(true); throw e; } this.afterSend(message); diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests.java index 662c13b32f..23c885407b 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests.java @@ -20,6 +20,8 @@ import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNotSame; import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; @@ -27,6 +29,10 @@ import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import java.io.IOException; +import java.io.OutputStream; +import java.net.Socket; +import java.nio.ByteBuffer; +import java.nio.channels.SocketChannel; import java.util.ArrayList; import java.util.List; import java.util.concurrent.Executors; @@ -39,12 +45,16 @@ import org.mockito.Mockito; import org.mockito.invocation.InvocationOnMock; import org.mockito.stubbing.Answer; +import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.ApplicationEvent; +import org.springframework.context.ApplicationEventPublisher; import org.springframework.integration.Message; import org.springframework.integration.MessagingException; import org.springframework.integration.core.PollableChannel; import org.springframework.integration.core.SubscribableChannel; import org.springframework.integration.ip.IpHeaders; +import org.springframework.integration.ip.tcp.serializer.ByteArrayCrLfSerializer; import org.springframework.integration.ip.util.TestingUtilities; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.support.MessageBuilder; @@ -286,6 +296,84 @@ public class CachingClientConnectionFactoryTests { verify(mockConn2).close(); } + @Test + public void testExceptionOnSendNet() throws Exception { + TcpConnectionSupport conn1 = mockedTcpNetConnection(); + TcpConnectionSupport conn2 = mockedTcpNetConnection(); + + CachingClientConnectionFactory cccf = createCCCFWith2Connections(conn1, conn2); + doTestCloseOnSendError(conn1, conn2, cccf); + } + + @Test + public void testExceptionOnSendNio() throws Exception { + TcpConnectionSupport conn1 = mockedTcpNioConnection(); + TcpConnectionSupport conn2 = mockedTcpNioConnection(); + + CachingClientConnectionFactory cccf = createCCCFWith2Connections(conn1, conn2); + doTestCloseOnSendError(conn1, conn2, cccf); + } + + private void doTestCloseOnSendError(TcpConnection conn1, TcpConnection conn2, + CachingClientConnectionFactory cccf) throws Exception { + TcpConnection cached1 = cccf.getConnection(); + try { + cached1.send(new GenericMessage("foo")); + fail("Expected IOException"); + } + catch (IOException e) { + assertEquals("Foo", e.getMessage()); + } + // Before INT-3163 this failed with a timeout - connection not returned to pool after failure on send() + TcpConnection cached2 = cccf.getConnection(); + assertTrue(cached1.getConnectionId().contains(conn1.getConnectionId())); + assertTrue(cached2.getConnectionId().contains(conn2.getConnectionId())); + } + + private CachingClientConnectionFactory createCCCFWith2Connections(TcpConnectionSupport conn1, TcpConnectionSupport conn2) + throws Exception { + AbstractClientConnectionFactory factory = mock(AbstractClientConnectionFactory.class); + when(factory.isRunning()).thenReturn(true); + when(factory.getConnection()).thenReturn(conn1, conn2); + CachingClientConnectionFactory cccf = new CachingClientConnectionFactory(factory, 1); + cccf.setConnectionWaitTimeout(1); + cccf.start(); + return cccf; + } + + private TcpConnectionSupport mockedTcpNetConnection() throws IOException { + Socket socket = mock(Socket.class); + when(socket.isClosed()).thenReturn(true); // closed when next retrieved + OutputStream stream = mock(OutputStream.class); + doThrow(new IOException("Foo")).when(stream).write(Mockito.any(byte[].class)); + when(socket.getOutputStream()).thenReturn(stream); + TcpNetConnection conn = new TcpNetConnection(socket, false, false, new ApplicationEventPublisher() { + + @Override + public void publishEvent(ApplicationEvent event) { + } + }, "foo"); + conn.setMapper(new TcpMessageMapper()); + conn.setSerializer(new ByteArrayCrLfSerializer()); + return conn; + } + + private TcpConnectionSupport mockedTcpNioConnection() throws Exception { + SocketChannel socketChannel = mock(SocketChannel.class); + new DirectFieldAccessor(socketChannel).setPropertyValue("open", false); + doThrow(new IOException("Foo")).when(socketChannel).write(Mockito.any(ByteBuffer.class)); + when(socketChannel.socket()).thenReturn(mock(Socket.class)); + TcpNioConnection conn = new TcpNioConnection(socketChannel, false, false, new ApplicationEventPublisher() { + + @Override + public void publishEvent(ApplicationEvent event) { + } + }, "foo"); + conn.setMapper(new TcpMessageMapper()); + conn.setSerializer(new ByteArrayCrLfSerializer()); + return conn; + } + private TcpConnectionSupport makeMockConnection(String name) { return makeMockConnection(name, false); } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionEventTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionEventTests.java index 24e2f0340f..7bfe7b1735 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionEventTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionEventTests.java @@ -24,7 +24,8 @@ import static org.mockito.Mockito.mock; import java.io.OutputStream; import java.net.Socket; -import java.util.concurrent.atomic.AtomicReference; +import java.util.ArrayList; +import java.util.List; import org.junit.Test; import org.mockito.Mockito; @@ -44,16 +45,17 @@ public class ConnectionEventTests { @Test public void test() throws Exception { Socket socket = mock(Socket.class); - final AtomicReference theEvent = new AtomicReference(); + final List theEvent = new ArrayList(); TcpNetConnection conn = new TcpNetConnection(socket, false, false, new ApplicationEventPublisher() { public void publishEvent(ApplicationEvent event) { - theEvent.set((TcpConnectionEvent) event); + theEvent.add((TcpConnectionEvent) event); } }, "foo"); - assertNotNull(theEvent.get()); - assertTrue(theEvent.get() instanceof TcpConnectionOpenEvent); - assertTrue(theEvent.get().toString().endsWith("[factory=foo, connectionId=" + conn.getConnectionId() + "] **OPENED**")); + assertTrue(theEvent.size() > 0); + assertNotNull(theEvent.get(0)); + assertTrue(theEvent.get(0) instanceof TcpConnectionOpenEvent); + assertTrue(theEvent.get(0).toString().endsWith("[factory=foo, connectionId=" + conn.getConnectionId() + "] **OPENED**")); @SuppressWarnings("unchecked") Serializer serializer = mock(Serializer.class); RuntimeException toBeThrown = new RuntimeException("foo"); @@ -65,16 +67,17 @@ public class ConnectionEventTests { fail("Expected exception"); } catch (Exception e) {} - assertNotNull(theEvent.get()); - assertTrue(theEvent.get() instanceof TcpConnectionExceptionEvent); - assertTrue(theEvent.get().toString().endsWith("[factory=foo, connectionId=" + conn.getConnectionId() + "]")); - assertTrue(theEvent.get().toString().contains("cause=java.lang.RuntimeException: foo]")); - TcpConnectionExceptionEvent event = (TcpConnectionExceptionEvent) theEvent.get(); + assertTrue(theEvent.size() > 1); + assertNotNull(theEvent.get(1)); + assertTrue(theEvent.get(1) instanceof TcpConnectionExceptionEvent); + assertTrue(theEvent.get(1).toString().endsWith("[factory=foo, connectionId=" + conn.getConnectionId() + "]")); + assertTrue(theEvent.get(1).toString().contains("cause=java.lang.RuntimeException: foo]")); + TcpConnectionExceptionEvent event = (TcpConnectionExceptionEvent) theEvent.get(1); assertNotNull(event.getCause()); assertSame(toBeThrown, event.getCause()); - conn.close(); - assertNotNull(theEvent.get()); - assertTrue(theEvent.get().toString().endsWith("[factory=foo, connectionId=" + conn.getConnectionId() + "] **CLOSED**")); + assertTrue(theEvent.size() > 2); + assertNotNull(theEvent.get(2)); + assertTrue(theEvent.get(2).toString().endsWith("[factory=foo, connectionId=" + conn.getConnectionId() + "] **CLOSED**")); } }