diff --git a/spring-integration-ip/.springBeans b/spring-integration-ip/.springBeans index 85a1da38fa..4d34282f88 100644 --- a/spring-integration-ip/.springBeans +++ b/spring-integration-ip/.springBeans @@ -1,13 +1,14 @@ - - - 1 - - - - - - - - - - + + + 1 + + + + + + + src/test/java/org/springframework/integration/ip/tcp/connection/SOLingerTests-context.xml + + + + 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 02d7b2b9bc..59884c6c71 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,7 @@ public class TcpSendingMessageHandler extends AbstractMessageHandler implements public void handleMessageInternal(final Message message) throws MessageRejectedException, MessageHandlingException, MessageDeliveryException { if (this.serverConnectionFactory != null) { - // We don't own the connection + // We don't own the connection, we are asynchronously replying Object connectionId = message.getHeaders().get(IpHeaders.CONNECTION_ID); TcpConnection connection = connections.get(connectionId); if (connection != null) { @@ -97,7 +97,7 @@ public class TcpSendingMessageHandler extends AbstractMessageHandler implements return; } - + // we own the connection try { doWrite(message); } catch (MessageMappingException e) { 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 627243abb2..ace9c238f5 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 @@ -60,11 +60,11 @@ public abstract class AbstractConnectionFactory private boolean soTcpNoDelay; - private int soLinger; + private int soLinger = -1; // don't set by default private boolean soKeepAlive; - private int soTrafficClass; + private int soTrafficClass = -1; // don't set by default protected Executor taskExecutor; diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnection.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnection.java index 10496fe9e0..142f5c24b2 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnection.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnection.java @@ -16,13 +16,15 @@ package org.springframework.integration.ip.tcp.connection; +import java.net.Socket; +import java.net.SocketException; import java.util.concurrent.atomic.AtomicLong; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; - import org.springframework.core.serializer.Deserializer; import org.springframework.core.serializer.Serializer; +import org.springframework.integration.Message; import org.springframework.integration.ip.tcp.serializer.AbstractByteArraySerializer; import org.springframework.util.Assert; @@ -48,6 +50,8 @@ public abstract class AbstractTcpConnection implements TcpConnection { protected TcpMessageMapper mapper; protected TcpListener listener; + + private TcpListener actualListener; protected TcpSender sender; @@ -59,8 +63,27 @@ public abstract class AbstractTcpConnection implements TcpConnection { private AtomicLong sequence = new AtomicLong(); - public AbstractTcpConnection(boolean server) { + private int soLinger = -1; + + public AbstractTcpConnection(Socket socket, boolean server) { this.server = server; + try { + this.soLinger = socket.getSoLinger(); + } catch (SocketException e) { } + } + + public void afterSend(Message message) throws Exception { + if (logger.isDebugEnabled()) + logger.debug("Message sent " + message); + if (this.singleUse) { + // if (we're a server socket, or a send-only socket), and soLinger <> 0, close + if ((this.isServer() || this.actualListener == null) && this.soLinger != 0) { + if (logger.isDebugEnabled()) { + logger.debug("Closing single-use connection" + this.getConnectionId()); + } + this.closeConnection(); + } + } } /** @@ -71,7 +94,23 @@ public abstract class AbstractTcpConnection implements TcpConnection { this.sender.removeDeadConnection(this); } } - + + /** + * If we have been intercepted, propagate the close from the outermost interceptor; + * otherwise, just call close(). + */ + protected void closeConnection() { + if (!(this.listener instanceof TcpConnectionInterceptor)) { + close(); + return; + } + TcpConnectionInterceptor outerInterceptor = (TcpConnectionInterceptor) this.listener; + while (outerInterceptor.getListener() instanceof TcpConnectionInterceptor) { + outerInterceptor = (TcpConnectionInterceptor) outerInterceptor.getListener(); + } + outerInterceptor.close(); + } + /** * @return the mapper */ @@ -129,6 +168,16 @@ public abstract class AbstractTcpConnection implements TcpConnection { */ public void registerListener(TcpListener listener) { this.listener = listener; + // Determine the actual listener for this connection + if (!(this.listener instanceof TcpConnectionInterceptor)) { + this.actualListener = this.listener; + } else { + TcpConnectionInterceptor outerInterceptor = (TcpConnectionInterceptor) this.listener; + while (outerInterceptor.getListener() instanceof TcpConnectionInterceptor) { + outerInterceptor = (TcpConnectionInterceptor) outerInterceptor.getListener(); + } + this.actualListener = outerInterceptor.getListener(); + } } /** diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnectionInterceptor.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnectionInterceptor.java index bbbadc186a..b4ef4b2a3a 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnectionInterceptor.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnectionInterceptor.java @@ -34,7 +34,8 @@ public abstract class AbstractTcpConnectionInterceptor implements TcpConnectionI private TcpListener tcpListener; private TcpSender tcpSender; - + + private Boolean realSender; public void close() { this.theConnection.close(); @@ -61,8 +62,8 @@ public abstract class AbstractTcpConnectionInterceptor implements TcpConnectionI } public void registerListener(TcpListener listener) { - this.theConnection.registerListener(this); this.tcpListener = listener; + this.theConnection.registerListener(this); } public void registerSender(TcpSender sender) { @@ -159,6 +160,21 @@ public abstract class AbstractTcpConnectionInterceptor implements TcpConnectionI public long getConnectionSeq() { return this.theConnection.getConnectionSeq(); } - + + TcpSender getSender() { + return this.tcpSender; + } + + protected boolean hasRealSender() { + if (this.realSender != null) { + return this.realSender; + } + TcpSender sender = this.getSender(); + while (sender != null && sender instanceof AbstractTcpConnectionInterceptor) { + sender = ((AbstractTcpConnectionInterceptor) sender).getSender(); + } + this.realSender = sender != null; + return this.realSender; + } } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptorFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptorFactory.java index a256e1dad8..6a69543134 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptorFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptorFactory.java @@ -26,8 +26,8 @@ package org.springframework.integration.ip.tcp.connection; public interface TcpConnectionInterceptorFactory { /** - * Called for each new connection - if an interceptor is - * stateful, a new interceptor must be returned on each call. + * Called for each new connection; + * a new interceptor must be returned on each call. * * @return the TcpInterceptor */ 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 88bb07a6af..ee3dc6b8d4 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 @@ -35,6 +35,8 @@ public class TcpNetConnection extends AbstractTcpConnection { private final Socket socket; + private boolean noReadErrorOnClose; + /** * Constructs a TcpNetConnection for the socket. * @param socket the socket @@ -42,7 +44,7 @@ public class TcpNetConnection extends AbstractTcpConnection { * a result of an incoming request. */ public TcpNetConnection(Socket socket, boolean server) { - super(server); + super(socket, server); this.socket = socket; getConnectionId(); } @@ -51,6 +53,7 @@ public class TcpNetConnection extends AbstractTcpConnection { * Closes this connection. */ public void close() { + this.noReadErrorOnClose = true; try { this.socket.close(); } catch (Exception e) {} @@ -65,8 +68,7 @@ public class TcpNetConnection extends AbstractTcpConnection { public synchronized void send(Message message) throws Exception { Object object = mapper.fromMessage(message); this.serializer.serialize(object, this.socket.getOutputStream()); - if (logger.isDebugEnabled()) - logger.debug("Message sent " + message); + this.afterSend(message); } public String getHostAddress() { @@ -108,12 +110,19 @@ public class TcpNetConnection extends AbstractTcpConnection { try { message = this.mapper.toMessage(this); } catch (Exception e) { - this.close(); + this.closeConnection(); if (!(e instanceof SoftEndOfStreamException)) { if (e instanceof SocketTimeoutException && this.singleUse) { logger.debug("Closing single use socket after timeout"); } else { - if (logger.isTraceEnabled()) { + if (this.noReadErrorOnClose) { + if (logger.isDebugEnabled()) { + logger.debug("Read exception " + + this.getConnectionId() + " " + + e.getClass().getSimpleName() + + ":" + e.getCause() + ":" + e.getMessage()); + } + } else if (logger.isTraceEnabled()) { logger.error("Read exception " + this.getConnectionId(), e); } else { @@ -138,7 +147,7 @@ public class TcpNetConnection extends AbstractTcpConnection { if (e instanceof NoListenerException) { if (this.singleUse) { logger.debug("Closing single use socket after inbound message " + this.connectionId); - this.close(); + this.closeConnection(); okToRun = false; } else { logger.warn("Unexpected message - no inbound adapter registered with connection " + message); @@ -154,7 +163,7 @@ public class TcpNetConnection extends AbstractTcpConnection { */ if (this.singleUse && ((!this.server && !intercepted) || (this.server && this.sender == null))) { logger.debug("Closing single use socket after inbound message " + this.connectionId); - this.close(); + this.closeConnection(); okToRun = false; } } 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 ccbd4dadd0..c4d159873f 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 @@ -71,7 +71,7 @@ public class TcpNioConnection extends AbstractTcpConnection { * a result of an incoming request. */ public TcpNioConnection(SocketChannel socketChannel, boolean server) throws Exception { - super(server); + super(socketChannel.socket(), server); this.socketChannel = socketChannel; this.pipedInputStream = new PipedInputStream(); this.pipedOutputStream = new PipedOutputStream(this.pipedInputStream); @@ -107,6 +107,7 @@ public class TcpNioConnection extends AbstractTcpConnection { synchronized(mapper) { Object object = mapper.fromMessage(message); this.serializer.serialize(object, this.channelOutputStream); + this.afterSend(message); } } @@ -209,7 +210,7 @@ public class TcpNioConnection extends AbstractTcpConnection { try { message = this.mapper.toMessage(this); } catch (Exception e) { - this.close(); + this.closeConnection(); if (e instanceof SocketTimeoutException && this.singleUse) { if (logger.isDebugEnabled()) { logger.debug("Closing single use socket after timeout " + this.connectionId); @@ -244,7 +245,7 @@ public class TcpNioConnection extends AbstractTcpConnection { if (logger.isDebugEnabled()) { logger.debug("Closing single use channel after inbound message " + this.connectionId); } - this.close(); + this.closeConnection(); } } else { logger.error("Exception sending meeeage: " + message, e); @@ -257,7 +258,7 @@ public class TcpNioConnection extends AbstractTcpConnection { */ if (this.singleUse && ((!this.server && !intercepted) || (this.server && this.sender == null))) { logger.debug("Closing single use cbannel after inbound message " + this.connectionId); - this.close(); + this.closeConnection(); } } @@ -275,7 +276,7 @@ public class TcpNioConnection extends AbstractTcpConnection { rawBuffer.clear(); int len = socketChannel.read(rawBuffer); if (len < 0) { - this.close(); + this.closeConnection(); throw new IOException("Channel closed"); } rawBuffer.flip(); @@ -310,7 +311,7 @@ public class TcpNioConnection extends AbstractTcpConnection { logger.error("Exception on Read " + this.getConnectionId() + " " + e.getMessage()); - this.close(); + this.closeConnection(); } } @@ -318,7 +319,7 @@ public class TcpNioConnection extends AbstractTcpConnection { * Close the socket due to timeout. */ void timeout() { - this.close(); + this.closeConnection(); } /** diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpInboundGatewayTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpInboundGatewayTests.java index 3d36a5a557..e4c92e41b8 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpInboundGatewayTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpInboundGatewayTests.java @@ -69,15 +69,16 @@ public class TcpInboundGatewayTests { return channel; } }); - Socket socket = SocketFactory.getDefault().createSocket("localhost", port); - socket.getOutputStream().write("Test1\r\n".getBytes()); - socket.getOutputStream().write("Test2\r\n".getBytes()); - handler.handleMessage(channel.receive()); - handler.handleMessage(channel.receive()); + Socket socket1 = SocketFactory.getDefault().createSocket("localhost", port); + socket1.getOutputStream().write("Test1\r\n".getBytes()); + Socket socket2 = SocketFactory.getDefault().createSocket("localhost", port); + socket2.getOutputStream().write("Test2\r\n".getBytes()); + handler.handleMessage(channel.receive(1000)); + handler.handleMessage(channel.receive(1000)); byte[] bytes = new byte[12]; - readFully(socket.getInputStream(), bytes); + readFully(socket1.getInputStream(), bytes); assertEquals("Echo:Test1\r\n", new String(bytes)); - readFully(socket.getInputStream(), bytes); + readFully(socket2.getInputStream(), bytes); assertEquals("Echo:Test2\r\n", new String(bytes)); } @@ -134,19 +135,17 @@ public class TcpInboundGatewayTests { return channel; } }); - Socket socket = SocketFactory.getDefault().createSocket("localhost", port); - socket.getOutputStream().write("Test1\r\n".getBytes()); - socket.getOutputStream().write("Test2\r\n".getBytes()); + Socket socket1 = SocketFactory.getDefault().createSocket("localhost", port); + socket1.getOutputStream().write("Test1\r\n".getBytes()); + Socket socket2 = SocketFactory.getDefault().createSocket("localhost", port); + socket2.getOutputStream().write("Test2\r\n".getBytes()); handler.handleMessage(channel.receive()); handler.handleMessage(channel.receive()); - Set results = new HashSet(); byte[] bytes = new byte[12]; - readFully(socket.getInputStream(), bytes); - results.add(new String(bytes)); - readFully(socket.getInputStream(), bytes); - results.add(new String(bytes)); - assertTrue(results.remove("Echo:Test1\r\n")); - assertTrue(results.remove("Echo:Test2\r\n")); + readFully(socket1.getInputStream(), bytes); + assertEquals("Echo:Test1\r\n", new String(bytes)); + readFully(socket2.getInputStream(), bytes); + assertEquals("Echo:Test2\r\n", new String(bytes)); } @Test @@ -210,13 +209,14 @@ public class TcpInboundGatewayTests { gateway.setRequestChannel(channel); ServiceActivatingHandler handler = new ServiceActivatingHandler(new FailingService()); channel.subscribe(handler); - Socket socket = SocketFactory.getDefault().createSocket("localhost", port); - socket.getOutputStream().write("Test1\r\n".getBytes()); - socket.getOutputStream().write("Test2\r\n".getBytes()); + Socket socket1 = SocketFactory.getDefault().createSocket("localhost", port); + socket1.getOutputStream().write("Test1\r\n".getBytes()); + Socket socket2 = SocketFactory.getDefault().createSocket("localhost", port); + socket2.getOutputStream().write("Test2\r\n".getBytes()); byte[] bytes = new byte[errorMessage.length() + 2]; - readFully(socket.getInputStream(), bytes); + readFully(socket1.getInputStream(), bytes); assertEquals(errorMessage + "\r\n", new String(bytes)); - readFully(socket.getInputStream(), bytes); + readFully(socket2.getInputStream(), bytes); assertEquals(errorMessage + "\r\n", new String(bytes)); } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java index 74c3c82ffa..f682ed66c2 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java @@ -518,6 +518,7 @@ public class TcpReceivingChannelAdapterTests { new ObjectOutputStream(socket.getOutputStream()).writeObject("Hello"); assertEquals("world!", new ObjectInputStream(socket.getInputStream()).readObject()); new ObjectOutputStream(socket.getOutputStream()).writeObject("Test1"); + socket = SocketFactory.getDefault().createSocket("localhost", port); new ObjectOutputStream(socket.getOutputStream()).writeObject("Hello"); assertEquals("world!", new ObjectInputStream(socket.getInputStream()).readObject()); diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/HelloWorldInterceptor.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/HelloWorldInterceptor.java index 67f602c564..33592964f3 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/HelloWorldInterceptor.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/HelloWorldInterceptor.java @@ -42,6 +42,8 @@ public class HelloWorldInterceptor extends AbstractTcpConnectionInterceptor { private String hello = "Hello"; private String world = "world!"; + + private boolean closeReceived; public HelloWorldInterceptor() { } @@ -86,24 +88,58 @@ public class HelloWorldInterceptor extends AbstractTcpConnectionInterceptor { return true; } } - return super.onMessage(message); + try { + return super.onMessage(message); + } finally { + // on the server side, we don't want to close if we are expecting a response + if (!(this.isServer() && this.hasRealSender())) { + this.checkDeferredClose(); + } + } } @Override public void send(Message message) throws Exception { - if (!this.negotiated) { - if (!this.isServer()) { - logger.debug("Sending " + hello); - super.send(MessageBuilder.withPayload(hello).build()); - this.negotiationSemaphore.tryAcquire(this.timeout, TimeUnit.MILLISECONDS); - if (!this.negotiated) { - throw new MessagingException("Negotiation error"); + try { + if (!this.negotiated) { + if (!this.isServer()) { + logger.debug("Sending " + hello); + super.send(MessageBuilder.withPayload(hello).build()); + this.negotiationSemaphore.tryAcquire(this.timeout, TimeUnit.MILLISECONDS); + if (!this.negotiated) { + throw new MessagingException("Negotiation error"); + } } } + super.send(message); + } finally { + this.checkDeferredClose(); } - super.send(message); } - + + /** + * Defer the close until we've actually sent the data after negotiation + */ + @Override + public void close() { + if (this.negotiated) { + super.close(); + return; + } + closeReceived = true; + logger.debug("Deferring close"); + } + + /** + * Execute the close, if deferred + */ + private void checkDeferredClose() { + if (this.closeReceived) { + logger.debug("Executing deferred close"); + this.close(); + } + } + } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/SOLingerTests-context.xml b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/SOLingerTests-context.xml new file mode 100644 index 0000000000..a9b95e5558 --- /dev/null +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/SOLingerTests-context.xml @@ -0,0 +1,83 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/SOLingerTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/SOLingerTests.java new file mode 100644 index 0000000000..695befd7df --- /dev/null +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/SOLingerTests.java @@ -0,0 +1,166 @@ +/* + * Copyright 2002-2010 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.ip.tcp.connection; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +import java.io.IOException; +import java.io.InputStream; +import java.net.Socket; +import java.net.SocketTimeoutException; + +import javax.net.SocketFactory; + +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Gary Russell + * @since 2.0.2 + * + */ +@RunWith(SpringJUnit4ClassRunner.class) +@ContextConfiguration +public class SOLingerTests { + + @Autowired + private AbstractServerConnectionFactory inCFNet; + + @Autowired + private AbstractServerConnectionFactory inCFNio; + + @Autowired + private AbstractServerConnectionFactory inCFNetRst; + + @Autowired + private AbstractServerConnectionFactory inCFNioRst; + + @Autowired + private AbstractServerConnectionFactory inCFNetLinger; + + @Autowired + private AbstractServerConnectionFactory inCFNioLinger; + + @Test + public void configOk() {} + + @Test + public void finReceivedNet() { + finReceived(inCFNet); + } + + @Test + public void finReceivedNio() { + finReceived(inCFNio); + } + + @Test + public void rstReceivedNet() { + rstReceived(inCFNetRst); + } + + @Test + public void rstReceivedNio() { + rstReceived(inCFNioRst); + } + + @Test + public void finReceivedNetLinger() { + finReceived(inCFNetLinger); + } + + @Test + public void finReceivedNioLinger() { + finReceived(inCFNioLinger); + } + + private void finReceived(AbstractServerConnectionFactory inCF) { + int port = inCF.getPort(); + int n = 0; + while (!inCF.isListening()) { + try { + Thread.sleep(100); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + fail("Interrupted"); + } + if (n++ > 100) { + fail("Failed to start"); + } + } + try { + Socket socket = SocketFactory.getDefault().createSocket("localhost", port); + String test = "Test\r\n"; + socket.getOutputStream().write(test.getBytes()); + byte[] buff = new byte[test.length() + 5]; + readFully(socket.getInputStream(), buff); + assertEquals("echo:" + test, new String(buff)); + n = socket.getInputStream().read(); + // we expect an orderly close + assertEquals(-1, n); + } catch (Exception e) { + e.printStackTrace(); + fail("Unexpected Exception " + e.getMessage()); + } + + } + + private void rstReceived(AbstractServerConnectionFactory inCF) { + int port = inCF.getPort(); + int n = 0; + while (!inCF.isListening()) { + try { + Thread.sleep(100); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + fail("Interrupted"); + } + if (n++ > 100) { + fail("Failed to start"); + } + } + try { + Socket socket = SocketFactory.getDefault().createSocket("localhost", port); + socket.setSoTimeout(200); + String test = "Test\r\n"; + socket.getOutputStream().write(test.getBytes()); + byte[] buff = new byte[test.length() + 5]; + readFully(socket.getInputStream(), buff); + assertEquals("echo:" + test, new String(buff)); + try { + n = socket.getInputStream().read(); + fail("Expected IOException"); + } catch (IOException ioe) { + assertTrue(ioe instanceof SocketTimeoutException); + } + } catch (Exception e) { + e.printStackTrace(); + fail("Unexpected Exception " + e.getMessage()); + } + + } + private void readFully(InputStream is, byte[] buff) throws IOException { + for (int i = 0; i < buff.length; i++) { + buff[i] = (byte) is.read(); + } + } + +} diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapperTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapperTests.java index 76f2e66fb0..414bdf62c0 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapperTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapperTests.java @@ -19,6 +19,10 @@ import static org.junit.Assert.assertEquals; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; +import java.net.Socket; + +import javax.net.SocketFactory; + import org.junit.Test; import org.springframework.integration.Message; import org.springframework.integration.ip.IpHeaders; @@ -68,7 +72,8 @@ public class TcpMessageMapperTests { public void testToMessageSequence() throws Exception { TcpMessageMapper mapper = new TcpMessageMapper(); - TcpConnection connection = new AbstractTcpConnection(false) { + Socket socket = SocketFactory.getDefault().createSocket(); + TcpConnection connection = new AbstractTcpConnection(socket, false) { public void run() { } public void send(Message message) throws Exception { diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionReadTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionReadTests.java index 502213ed7c..8a27d3ea56 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionReadTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionReadTests.java @@ -235,6 +235,7 @@ public class TcpNioConnectionReadTests { SocketTestUtils.testSendLengthOverflow(port); whileOpen(semaphore, added); assertEquals(1, added.size()); + assertTrue(semaphore.tryAcquire(10000, TimeUnit.MILLISECONDS)); assertTrue(removed.size() > 0); scf.close(); } @@ -273,6 +274,7 @@ public class TcpNioConnectionReadTests { SocketTestUtils.testSendStxEtxOverflow(port); whileOpen(semaphore, added); assertEquals(1, added.size()); + assertTrue(semaphore.tryAcquire(10000, TimeUnit.MILLISECONDS)); assertTrue(removed.size() > 0); scf.close(); } @@ -311,6 +313,7 @@ public class TcpNioConnectionReadTests { SocketTestUtils.testSendCrLfOverflow(port); whileOpen(semaphore, added); assertEquals(1, added.size()); + assertTrue(semaphore.tryAcquire(10000, TimeUnit.MILLISECONDS)); assertTrue(removed.size() > 0); scf.close(); } @@ -349,6 +352,7 @@ public class TcpNioConnectionReadTests { socket.close(); whileOpen(semaphore, added); assertEquals(1, added.size()); + assertTrue(semaphore.tryAcquire(10000, TimeUnit.MILLISECONDS)); assertTrue(removed.size() > 0); scf.close(); } @@ -416,6 +420,7 @@ public class TcpNioConnectionReadTests { socket.close(); whileOpen(semaphore, added); assertEquals(1, added.size()); + assertTrue(semaphore.tryAcquire(10000, TimeUnit.MILLISECONDS)); assertTrue(removed.size() > 0); scf.close(); }