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 5e54612689..1309caed33 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 @@ -32,6 +32,7 @@ import org.springframework.integration.ip.tcp.connection.ClientModeConnectionMan import org.springframework.integration.ip.tcp.connection.TcpConnection; import org.springframework.integration.ip.tcp.connection.TcpListener; import org.springframework.integration.ip.tcp.connection.TcpSender; +import org.springframework.integration.message.ErrorMessage; import org.springframework.util.Assert; /** @@ -53,7 +54,7 @@ public class TcpInboundGateway extends MessagingGatewaySupport implements private volatile AbstractClientConnectionFactory clientConnectionFactory; - private Map connections = new ConcurrentHashMap(); + private final Map connections = new ConcurrentHashMap(); private volatile boolean isClientMode; @@ -76,6 +77,13 @@ public class TcpInboundGateway extends MessagingGatewaySupport implements } } else { + if (message instanceof ErrorMessage) { + /* + * Socket errors are sent here so they can be conveyed to any waiting thread. + * There's not one here; simply ignore. + */ + return false; + } this.activeCount.incrementAndGet(); try { return doOnMessage(message); diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java index cdf408938e..1f0d005e98 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java @@ -33,7 +33,9 @@ import org.springframework.integration.ip.tcp.connection.AbstractClientConnectio import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory; import org.springframework.integration.ip.tcp.connection.TcpConnection; import org.springframework.integration.ip.tcp.connection.TcpListener; +import org.springframework.integration.ip.tcp.connection.TcpNioClientConnectionFactory; import org.springframework.integration.ip.tcp.connection.TcpSender; +import org.springframework.integration.message.ErrorMessage; import org.springframework.util.Assert; /** @@ -163,8 +165,17 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler imp } AsyncReply reply = pendingReplies.get(connectionId); if (reply == null) { - logger.error("Cannot correlate response - no pending reply"); - return false; + if (message instanceof ErrorMessage) { + /* + * Socket errors are sent here so they can be conveyed to any waiting thread. + * If there's not one, simply ignore. + */ + return false; + } + else { + logger.error("Cannot correlate response - no pending reply"); + return false; + } } reply.setReply(message); return false; @@ -248,10 +259,13 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler imp private final CountDownLatch latch; + private final CountDownLatch secondChanceLatch; + private volatile Message reply; public AsyncReply() { this.latch = new CountDownLatch(1); + this.secondChanceLatch = new CountDownLatch(1); } /** @@ -268,12 +282,41 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler imp catch (InterruptedException e) { Thread.currentThread().interrupt(); } + boolean waitForMessageAfterError = connectionFactory instanceof TcpNioClientConnectionFactory; + while (reply instanceof ErrorMessage) { + if (waitForMessageAfterError) { + /* + * Possible race condition with NIO; we might have received the close + * before the reply, on a different thread. + */ + logger.debug("second chance"); + this.secondChanceLatch.await(2, TimeUnit.SECONDS); + waitForMessageAfterError = false; + } + else if (reply.getPayload() instanceof MessagingException) { + throw (MessagingException) reply.getPayload(); + } + else { + throw new MessagingException("Exception while awaiting reply", (Throwable) reply.getPayload()); + } + } return this.reply; } + /** + * We have a race condition when a socket is closed right after the reply is received. The close "error" + * might arrive before the actual reply. Overwrite an error with a good reply, but not vice-versa. + * @param reply + */ public void setReply(Message reply) { - this.reply = reply; - this.latch.countDown(); + if (this.reply == null) { + this.reply = reply; + this.latch.countDown(); + } + else if (this.reply instanceof ErrorMessage) { + this.reply = reply; + this.secondChanceLatch.countDown(); + } } } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapter.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapter.java index 6b0506c229..61e607dcbb 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapter.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapter.java @@ -28,6 +28,7 @@ import org.springframework.integration.ip.tcp.connection.ClientModeCapable; import org.springframework.integration.ip.tcp.connection.ClientModeConnectionManager; import org.springframework.integration.ip.tcp.connection.ConnectionFactory; import org.springframework.integration.ip.tcp.connection.TcpListener; +import org.springframework.integration.message.ErrorMessage; import org.springframework.util.Assert; /** @@ -68,6 +69,13 @@ public class TcpReceivingChannelAdapter } } else { + if (message instanceof ErrorMessage) { + /* + * Socket errors are sent here so they can be conveyed to any waiting thread. + * There's not one here; simply ignore. + */ + return false; + } this.activeCount.incrementAndGet(); try { sendMessage(message); diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java index 25b11d562c..251d952c41 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java @@ -16,6 +16,7 @@ package org.springframework.integration.ip.tcp.connection; +import java.io.EOFException; import java.io.IOException; import java.net.Socket; import java.net.SocketException; @@ -597,6 +598,9 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport key.interestOps(SelectionKey.OP_READ); selector.wakeup(); } + else { + connection.sendExceptionToListener(new EOFException("Connection is closed")); + } }}); } else if (key.isAcceptable()) { diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptorSupport.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptorSupport.java index 3a988fa4cc..e46ef42ec5 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptorSupport.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptorSupport.java @@ -19,6 +19,7 @@ package org.springframework.integration.ip.tcp.connection; import org.springframework.core.serializer.Deserializer; import org.springframework.core.serializer.Serializer; import org.springframework.integration.Message; +import org.springframework.integration.message.ErrorMessage; /** * Base class for TcpConnectionIntercepters; passes all method calls through @@ -131,7 +132,12 @@ public abstract class TcpConnectionInterceptorSupport extends TcpConnectionSuppo public boolean onMessage(Message message) { if (this.tcpListener == null) { - throw new NoListenerException("No listener registered for message reception"); + if (message instanceof ErrorMessage) { + return false; + } + else { + throw new NoListenerException("No listener registered for message reception"); + } } return this.tcpListener.onMessage(message); } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionSupport.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionSupport.java index f2c08d1b75..2b702091b6 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionSupport.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionSupport.java @@ -19,6 +19,8 @@ package org.springframework.integration.ip.tcp.connection; import java.net.InetAddress; import java.net.Socket; import java.net.SocketException; +import java.util.Collections; +import java.util.Map; import java.util.UUID; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; @@ -30,7 +32,9 @@ import org.springframework.context.ApplicationEventPublisher; import org.springframework.core.serializer.Deserializer; import org.springframework.core.serializer.Serializer; import org.springframework.integration.Message; +import org.springframework.integration.ip.IpHeaders; import org.springframework.integration.ip.tcp.serializer.AbstractByteArraySerializer; +import org.springframework.integration.message.ErrorMessage; import org.springframework.util.Assert; /** @@ -80,6 +84,8 @@ public abstract class TcpConnectionSupport implements TcpConnection { private final AtomicBoolean closePublished = new AtomicBoolean(); + private final AtomicBoolean exceptionSent = new AtomicBoolean(); + public TcpConnectionSupport() { this.server = false; this.applicationEventPublisher = null; @@ -302,6 +308,15 @@ public abstract class TcpConnectionSupport implements TcpConnection { return this.connectionId; } + protected final void sendExceptionToListener(Exception e) { + if (!this.exceptionSent.getAndSet(true) && this.getListener() != null) { + Map headers = Collections.singletonMap(IpHeaders.CONNECTION_ID, + (Object) this.getConnectionId()); + ErrorMessage errorMessage = new ErrorMessage(e, headers); + this.getListener().onMessage(errorMessage); + } + } + protected void publishConnectionOpenEvent() { TcpConnectionEvent event = new TcpConnectionOpenEvent(this, this.connectionFactoryName); 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 2472f9a747..fc5821d353 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 @@ -248,6 +248,7 @@ public class TcpNetConnection extends TcpConnectionSupport { e.getClass().getSimpleName() + ":" + (e.getCause() != null ? e.getCause() + ":" : "") + e.getMessage()); } + this.sendExceptionToListener(e); } } } 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 a4bf9a7f5f..75f2d3b455 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 @@ -201,23 +201,28 @@ public class TcpNioConnection extends TcpConnectionSupport { if (message != null) { sendToChannel(message); } - } else { + } + else { this.executionControl.decrementAndGet(); } - } catch (Exception e) { + } + catch (Exception e) { if (logger.isTraceEnabled()) { logger.error("Read exception " + this.getConnectionId(), e); - } else { + } + else { logger.error("Read exception " + this.getConnectionId() + " " + e.getClass().getSimpleName() + ":" + e.getCause() + ":" + e.getMessage()); } this.closeConnection(); + this.sendExceptionToListener(e); return; } - } finally { + } + finally { if (logger.isTraceEnabled()) { logger.trace(this.getConnectionId() + " Nio message assembler exiting..."); } @@ -228,7 +233,8 @@ public class TcpNioConnection extends TcpConnectionSupport { if (this.isOpen() && dataAvailable()) { checkForAssembler(); } - } catch (IOException e) { + } + catch (IOException e) { logger.error("Exception when checking for assembler", e); } } @@ -293,7 +299,7 @@ public class TcpNioConnection extends TcpConnectionSupport { */ if (this.isSingleUse() && ((!this.isServer() && !intercepted) || (this.isServer() && this.getSender() == null))) { if (logger.isDebugEnabled()) { - logger.debug("Closing single use cbannel after inbound message " + this.getConnectionId()); + logger.debug("Closing single use channel after inbound message " + this.getConnectionId()); } this.closeConnection(); } @@ -377,6 +383,7 @@ public class TcpNioConnection extends TcpConnectionSupport { // only execute run() if we don't already have one running this.executionControl.set(1); this.taskExecutor.execute(this); + logger.debug("Running an assembler"); } else { this.executionControl.decrementAndGet(); } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java index b386809392..518d611e74 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java @@ -18,6 +18,7 @@ package org.springframework.integration.ip.tcp; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; import static org.mockito.Mockito.doThrow; @@ -25,12 +26,14 @@ import static org.mockito.Mockito.mock; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; +import java.io.EOFException; import java.io.IOException; import java.io.ObjectInputStream; import java.io.ObjectOutputStream; import java.net.ServerSocket; import java.net.Socket; import java.util.ArrayList; +import java.util.Collections; import java.util.HashSet; import java.util.List; import java.util.Map; @@ -63,6 +66,7 @@ import org.springframework.integration.ip.tcp.connection.CachingClientConnection import org.springframework.integration.ip.tcp.connection.FailoverClientConnectionFactory; import org.springframework.integration.ip.tcp.connection.TcpConnectionSupport; import org.springframework.integration.ip.tcp.connection.TcpNetClientConnectionFactory; +import org.springframework.integration.ip.tcp.connection.TcpNioClientConnectionFactory; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.test.util.SocketUtils; @@ -87,6 +91,7 @@ public class TcpOutboundGatewayTests { try { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port, 100); latch.countDown(); + List sockets = new ArrayList(); int i = 0; while (true) { Socket socket = server.accept(); @@ -94,6 +99,7 @@ public class TcpOutboundGatewayTests { ois.readObject(); ObjectOutputStream oos = new ObjectOutputStream(socket.getOutputStream()); oos.writeObject("Reply" + (i++)); + sockets.add(socket); } } catch (Exception e) { if (!done.get()) { @@ -568,4 +574,116 @@ public class TcpOutboundGatewayTests { return connection; } + @Test + public void testNetGWPropagatesSocketClose() throws Exception { + final int port = SocketUtils.findAvailableServerSocket(); + AbstractClientConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port); + ccf.setSerializer(new DefaultSerializer()); + ccf.setDeserializer(new DefaultDeserializer()); + ccf.setSoTimeout(10000); + ccf.setSingleUse(false); + ccf.start(); + testGWPropagatesSocketCloseGuts(port, ccf); + } + + @Test + public void testNioGWPropagatesSocketClose() throws Exception { + final int port = SocketUtils.findAvailableServerSocket(); + AbstractClientConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port); + ccf.setSerializer(new DefaultSerializer()); + ccf.setDeserializer(new DefaultDeserializer()); + ccf.setSoTimeout(10000); + ccf.setSingleUse(false); + ccf.start(); + testGWPropagatesSocketCloseGuts(port, ccf); + } + + @Test + public void testCachedGWPropagatesSocketClose() throws Exception { + final int port = SocketUtils.findAvailableServerSocket(); + AbstractClientConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port); + ccf.setSerializer(new DefaultSerializer()); + ccf.setDeserializer(new DefaultDeserializer()); + ccf.setSoTimeout(10000); + ccf.setSingleUse(false); + CachingClientConnectionFactory cccf = new CachingClientConnectionFactory(ccf, 1); + cccf.start(); + testGWPropagatesSocketCloseGuts(port, cccf); + } + + @Test + public void testFailoverGWPropagatesSocketClose() throws Exception { + final int port = SocketUtils.findAvailableServerSocket(); + AbstractClientConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port); + ccf.setSerializer(new DefaultSerializer()); + ccf.setDeserializer(new DefaultDeserializer()); + ccf.setSoTimeout(10000); + ccf.setSingleUse(false); + FailoverClientConnectionFactory focf = new FailoverClientConnectionFactory( + Collections.singletonList(ccf)); + focf.start(); + testGWPropagatesSocketCloseGuts(port, focf); + } + + private void testGWPropagatesSocketCloseGuts(final int port, AbstractClientConnectionFactory ccf) throws Exception { + final CountDownLatch latch = new CountDownLatch(1); + final AtomicBoolean done = new AtomicBoolean(); + final AtomicReference lastReceived = new AtomicReference(); + final CountDownLatch serverLatch = new CountDownLatch(1); + + Executors.newSingleThreadExecutor().execute(new Runnable() { + + public void run() { + try { + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + latch.countDown(); + int i = 0; + while (!done.get()) { + Socket socket = server.accept(); + i++; + while (!socket.isClosed()) { + try { + ObjectInputStream ois = new ObjectInputStream(socket.getInputStream()); + String request = (String) ois.readObject(); + logger.debug("Read " + request + " closing socket"); + socket.close(); + lastReceived.set(request); + serverLatch.countDown(); + } + catch (IOException e) { + socket.close(); + } + } + } + } + catch (Exception e) { + if (!done.get()) { + e.printStackTrace(); + } + } + } + }); + assertTrue(latch.await(10000, TimeUnit.MILLISECONDS)); + final TcpOutboundGateway gateway = new TcpOutboundGateway(); + gateway.setConnectionFactory(ccf); + gateway.setRequestTimeout(Integer.MAX_VALUE); + QueueChannel replyChannel = new QueueChannel(); + gateway.setRequiresReply(true); + gateway.setOutputChannel(replyChannel); + gateway.setRemoteTimeout(5000); + gateway.afterPropertiesSet(); + gateway.start(); + try { + gateway.handleMessage(MessageBuilder.withPayload("Test").build()); + fail("expected failure"); + } + catch (Exception e) { + assertTrue(e.getCause() instanceof EOFException); + } + assertEquals(0, TestUtils.getPropertyValue(gateway, "pendingReplies", Map.class).size()); + Message reply = replyChannel.receive(0); + assertNull(reply); + done.set(true); + ccf.getConnection(); + } } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNetConnectionTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNetConnectionTests.java index d786b7a0a5..c0269c24e1 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNetConnectionTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNetConnectionTests.java @@ -30,7 +30,6 @@ import java.nio.channels.SocketChannel; import java.util.concurrent.atomic.AtomicReference; import org.apache.commons.logging.Log; - import org.junit.Test; import org.mockito.Mockito; import org.mockito.invocation.InvocationOnMock; @@ -43,6 +42,7 @@ import org.springframework.integration.Message; import org.springframework.integration.ip.tcp.connection.TcpNioConnection.ChannelInputStream; import org.springframework.integration.ip.tcp.serializer.ByteArrayStxEtxSerializer; import org.springframework.integration.ip.tcp.serializer.MapJsonSerializer; +import org.springframework.integration.message.ErrorMessage; import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.support.converter.MapMessageConverter; import org.springframework.integration.test.util.TestUtils; @@ -133,7 +133,9 @@ public class TcpNetConnectionTests { TcpListener listener = new TcpListener() { public boolean onMessage(Message message) { - inboundMessage.set(message); + if (!(message instanceof ErrorMessage)) { + inboundMessage.set(message); + } return false; } };