diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/SimpleTcpNetOutboundGateway.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/SimpleTcpNetOutboundGateway.java index d64ada4960..7dc5acae17 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/SimpleTcpNetOutboundGateway.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/SimpleTcpNetOutboundGateway.java @@ -86,6 +86,7 @@ public class SimpleTcpNetOutboundGateway extends return object; } catch (Exception e) { this.reader = null; + this.handler.close(); throw new MessagingException(requestMessage, e); } } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/SimpleTcpNetOutboundGatewayTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/SimpleTcpNetOutboundGatewayTests.java index fde11a4abb..a304b5fc51 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/SimpleTcpNetOutboundGatewayTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/SimpleTcpNetOutboundGatewayTests.java @@ -16,6 +16,7 @@ package org.springframework.integration.ip.tcp; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.fail; import java.net.ServerSocket; import java.net.Socket; @@ -187,4 +188,49 @@ public class SimpleTcpNetOutboundGatewayTests { assertEquals("OK", new String(bytes)); } + @Test + public void testOutboundCloseOnTimeout() throws Exception { + final int port = SocketUtils.findAvailableServerSocket(); + final CountDownLatch latch1 = new CountDownLatch(1); + final CountDownLatch latch2 = new CountDownLatch(1); + final CountDownLatch latch3 = new CountDownLatch(1); + Thread t = new Thread(new Runnable() { + public void run() { + try { + ServerSocket ss = ServerSocketFactory.getDefault().createServerSocket(port); + latch1.countDown(); + boolean first = true; + while (true) { + Socket s = ss.accept(); + byte[] b = new byte[1024]; + s.getInputStream().read(b); + if (!first) + s.getOutputStream().write("OK\r\n".getBytes()); + first = false; + latch2.countDown(); + latch3.await(); + s.close(); + } + } catch (Exception e) { + e.printStackTrace(); + } + }}); + t.start(); + latch1.await(2000, TimeUnit.MILLISECONDS); + SimpleTcpNetOutboundGateway gateway = new SimpleTcpNetOutboundGateway + ("localhost", port); + gateway.setMessageFormat(MessageFormats.FORMAT_CRLF); + gateway.setClose(false); + gateway.setSoTimeout(500); + Message message = MessageBuilder.withPayload("test").build(); + try { + gateway.handleRequestMessage(message); + fail("Expected failure"); + } catch (Exception e) { } + latch3.countDown(); + latch2.await(2000, TimeUnit.MILLISECONDS); + byte[] bytes = (byte[]) gateway.handleRequestMessage(message); + assertEquals("OK", new String(bytes)); + } + } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/util/SocketUtils.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/util/SocketUtils.java index ba9306c3ad..ed057a4ae1 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/util/SocketUtils.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/util/SocketUtils.java @@ -186,9 +186,7 @@ public class SocketUtils { writeByte(outputStream, 'x', true); } Thread.sleep(1000000000L); // wait forever, but we're a daemon - } catch (Exception e) { - e.printStackTrace(); - } + } catch (Exception e) { } } }); thread.setDaemon(true); @@ -293,9 +291,7 @@ public class SocketUtils { writeByte(outputStream, 'x', true); } Thread.sleep(1000000000L); // wait forever, but we're a daemon - } catch (Exception e) { - e.printStackTrace(); - } + } catch (Exception e) { } } }); thread.setDaemon(true);