diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolReceivingChannelAdapter.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolReceivingChannelAdapter.java index 5b730b78a8..6ce2f9dc26 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolReceivingChannelAdapter.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolReceivingChannelAdapter.java @@ -42,6 +42,8 @@ public abstract class AbstractInternetProtocolReceivingChannelAdapter protected volatile boolean active; + protected volatile boolean listening; + public AbstractInternetProtocolReceivingChannelAdapter(int port) { this.port = port; @@ -95,4 +97,8 @@ public abstract class AbstractInternetProtocolReceivingChannelAdapter this.active = false; } + public boolean isListening() { + return listening; + } + } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java index 79877c9c26..c5a901708e 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java @@ -87,6 +87,8 @@ public abstract class IpAdapterParserUtils { static final String SO_TRAFFIC_CLASS = "so-traffic-class"; + static final String CLOSE = "close"; + /** * Adds a constructor-arg to the provided bean definition builder @@ -172,6 +174,22 @@ public abstract class IpAdapterParserUtils { return multicast; } + /** + * Sets the close attribute, if present. + * @param element + */ + static void setClose(Element element, BeanDefinitionBuilder builder) { + String close = element.getAttribute(IpAdapterParserUtils.CLOSE); + if (!StringUtils.hasText(close)) { + close = "false"; + } + if (close.equals("true")) { + builder.addPropertyValue( + Conventions.attributeNameToPropertyName(IpAdapterParserUtils.CLOSE), + close); + } + } + /** * Gets the use-nio attribute, if present; if not returns 'false'. * @param element diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpInboundChannelAdapterParser.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpInboundChannelAdapterParser.java index 0f47d98388..49c7fa017b 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpInboundChannelAdapterParser.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpInboundChannelAdapterParser.java @@ -124,6 +124,7 @@ public class IpInboundChannelAdapterParser extends AbstractChannelAdapterParser IpAdapterParserUtils.USING_DIRECT_BUFFERS); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, IpAdapterParserUtils.SO_KEEP_ALIVE); + IpAdapterParserUtils.setClose(element, builder); return builder; } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpOutboundGatewayParser.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpOutboundGatewayParser.java index ac61531ff3..0178c6d655 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpOutboundGatewayParser.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpOutboundGatewayParser.java @@ -26,6 +26,7 @@ import org.w3c.dom.Element; * Parser for the <outbound-gateway> element of the integration 'jms' namespace. * * @author Gary Russell + * @since 2.0 */ public class IpOutboundGatewayParser extends AbstractConsumerEndpointParser { @@ -44,6 +45,7 @@ public class IpOutboundGatewayParser extends AbstractConsumerEndpointParser { IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, IpAdapterParserUtils.CUSTOM_SOCKET_READER_CLASS_NAME); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "reply-channel"); + IpAdapterParserUtils.setClose(element, builder); return builder; } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/AbstractSocketReader.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/AbstractSocketReader.java index 1bfe795ba1..813b0780a9 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/AbstractSocketReader.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/AbstractSocketReader.java @@ -33,6 +33,7 @@ import java.io.IOException; * when the format is {@link MessageFormats#FORMAT_CUSTOM}. * * @author Gary Russell + * @since 2.0 * */ public abstract class AbstractSocketReader implements SocketReader, MessageFormats { @@ -52,21 +53,21 @@ public abstract class AbstractSocketReader implements SocketReader, MessageForma * @return True when a message is completely assembled. * @throws IOException */ - protected abstract boolean assembleDataLengthFormat() throws IOException; + protected abstract int assembleDataLengthFormat() throws IOException; /** * Assembles data in format {@link #FORMAT_STX_ETX}. * @return True when a message is completely assembled. * @throws IOException */ - protected abstract boolean assembleDataStxEtxFormat() throws IOException; + protected abstract int assembleDataStxEtxFormat() throws IOException; /** * Assembles data in format {@link #FORMAT_CRLF}. * @return True when a message is completely assembled. * @throws IOException */ - protected abstract boolean assembleDataCrLfFormat() throws IOException; + protected abstract int assembleDataCrLfFormat() throws IOException; /** * Assembles data in format {@link #FORMAT_CUSTOM}. Implementations must @@ -76,9 +77,9 @@ public abstract class AbstractSocketReader implements SocketReader, MessageForma * @return True when a message is completely assembled. * @throws IOException */ - protected abstract boolean assembleDataCustomFormat() throws IOException; + protected abstract int assembleDataCustomFormat() throws IOException; - public boolean assembleData() throws IOException { + public int assembleData() throws IOException { try { switch (this.messageFormat) { case FORMAT_LENGTH_HEADER: diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/AbstractSocketWriter.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/AbstractSocketWriter.java index 417ce080fe..5ee036aa85 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/AbstractSocketWriter.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/AbstractSocketWriter.java @@ -17,6 +17,9 @@ package org.springframework.integration.ip.tcp; import java.io.IOException; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + /** * Abstract SocketWriter that handles data in 3 standard, and one custom * format. The default format is {@link MessageFormats#FORMAT_LENGTH_HEADER} in which @@ -31,14 +34,18 @@ import java.io.IOException; * appropriate implementation, and provide an implementation for * {@link #writeCustomFormat(byte[])} which is invoked by {@link #write(byte[])} * when the format is {@link MessageFormats#FORMAT_CUSTOM}. + * * @author Gary Russell + * @since 2.0 * */ public abstract class AbstractSocketWriter implements SocketWriter, MessageFormats { protected int messageFormat = FORMAT_LENGTH_HEADER; - /* + protected final Log logger = LogFactory.getLog(this.getClass()); + + /* * @see org.springframework.integration.ip.tcp.SocketWriter#write(byte[]) */ public synchronized void write(byte[] bytes) throws IOException { diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/AbstractTcpReceivingChannelAdapter.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/AbstractTcpReceivingChannelAdapter.java index 49bea661c7..02e68257b7 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/AbstractTcpReceivingChannelAdapter.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/AbstractTcpReceivingChannelAdapter.java @@ -29,6 +29,7 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; * are provided. * * @author Gary Russell + * @since 2.0 * */ public abstract class AbstractTcpReceivingChannelAdapter extends @@ -42,7 +43,9 @@ public abstract class AbstractTcpReceivingChannelAdapter extends protected volatile boolean soKeepAlive; - protected int messageFormat = MessageFormats.FORMAT_LENGTH_HEADER; + protected volatile int messageFormat = MessageFormats.FORMAT_LENGTH_HEADER; + + protected volatile boolean close; /** * Constructs a receiving channel adapter that listens on the port. @@ -122,4 +125,11 @@ public abstract class AbstractTcpReceivingChannelAdapter extends this.poolSize = poolSize; } + /** + * @param close the close to set + */ + public void setClose(boolean close) { + this.close = close; + } + } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/NetSocketReader.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/NetSocketReader.java index a109abc1e9..5810d2e5f3 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/NetSocketReader.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/NetSocketReader.java @@ -32,6 +32,7 @@ import org.springframework.integration.message.MessageMappingException; * is completely assembled. * * @author Gary Russell + * @since 2.0 * */ public class NetSocketReader extends AbstractSocketReader { @@ -52,9 +53,11 @@ public class NetSocketReader extends AbstractSocketReader { * @see org.springframework.integration.ip.tcp.SocketReader#read(java.nio.ByteBuffer) */ @Override - protected boolean assembleDataLengthFormat() throws IOException { + protected int assembleDataLengthFormat() throws IOException { byte[] lengthPart = new byte[4]; - read(lengthPart); + int status = read(lengthPart, true); + if (status < 0) + return status; int messageLength = ByteBuffer.wrap(lengthPart).getInt(); if (logger.isDebugEnabled()) { logger.debug("Message length is " + messageLength); @@ -64,27 +67,27 @@ public class NetSocketReader extends AbstractSocketReader { " exceeds max message length: " + this.maxMessageSize); } byte[] messagePart = new byte[messageLength]; - read(messagePart); + read(messagePart, false); assembledData = messagePart; - return true; + return MESSAGE_COMPLETE; } /* (non-Javadoc) * @see org.springframework.integration.ip.tcp.AbstractSocketReader#assembleDataStxEtxFormat() */ @Override - protected boolean assembleDataStxEtxFormat() throws IOException { + protected int assembleDataStxEtxFormat() throws IOException { InputStream inputStream = socket.getInputStream(); - if (inputStream.read() != STX) + int bite = inputStream.read(); + if (bite < 0) { + return bite; + } + if (bite != STX) throw new MessageMappingException("Expected STX to begin message"); byte[] buffer = new byte[this.maxMessageSize]; int n = 0; - int bite; while ((bite = inputStream.read()) != ETX) { - if (bite < 0) { - logger.debug("Socket closed"); - throw new IOException("Socket Closed"); - } + checkClosure(bite); buffer[n++] = (byte) bite; if (n >= this.maxMessageSize) { throw new IOException("ETX not found before max message length: " @@ -93,24 +96,31 @@ public class NetSocketReader extends AbstractSocketReader { } assembledData = new byte[n]; System.arraycopy(buffer, 0, assembledData, 0, n); - return true; + return MESSAGE_COMPLETE; + } + + private void checkClosure(int bite) throws IOException { + if (bite < 0) { + logger.debug("Socket closed"); + throw new IOException("Socket closed"); + } } /* (non-Javadoc) * @see org.springframework.integration.ip.tcp.AbstractSocketReader#assembleDataCrLfFormat() */ @Override - protected boolean assembleDataCrLfFormat() throws IOException { + protected int assembleDataCrLfFormat() throws IOException { InputStream inputStream = socket.getInputStream(); byte[] buffer = new byte[this.maxMessageSize]; int n = 0; int bite; while (true) { bite = inputStream.read(); - if (bite < 0) { - logger.debug("Socket closed"); - throw new IOException("Socket Closed"); + if (bite < 0 && n == 0) { + return bite; } + checkClosure(bite); if (n > 0 && bite == '\n' && buffer[n-1] == '\r') break; buffer[n++] = (byte) bite; @@ -121,7 +131,7 @@ public class NetSocketReader extends AbstractSocketReader { }; assembledData = new byte[n-1]; System.arraycopy(buffer, 0, assembledData, 0, n-1); - return true; + return MESSAGE_COMPLETE; } /** @@ -132,7 +142,7 @@ public class NetSocketReader extends AbstractSocketReader { * */ @Override - protected boolean assembleDataCustomFormat() throws IOException { + protected int assembleDataCustomFormat() throws IOException { throw new UnsupportedOperationException("Need to subclass for this format"); } @@ -149,19 +159,23 @@ public class NetSocketReader extends AbstractSocketReader { * Reads data from the socket and puts the data in buffer. Blocks until * buffer is full or a socket timeout occurs. * @param buffer + * @param header true if we are reading the header + * @return < 0 if socket closed and not in the middle of a message * @throws IOException */ - protected void read(byte[] buffer) throws IOException { + protected int read(byte[] buffer, boolean header) throws IOException { int lengthRead = 0; int needed = buffer.length; while (lengthRead < needed) { int len; len = socket.getInputStream().read(buffer, lengthRead, needed - lengthRead); - if (len < 0) { - logger.debug("Socket closed"); - throw new IOException("Socket Closed"); + if (len < 0 && header && lengthRead == 0) { + return len; } + if (len < 0) + logger.debug("socket closed after " + lengthRead + " of " + needed); + checkClosure(len); lengthRead += len; if (logger.isDebugEnabled()) { logger.debug("Read " + len + " bytes, buffer is now at " + @@ -169,7 +183,7 @@ public class NetSocketReader extends AbstractSocketReader { needed); } } - + return 0; } /* (non-Javadoc) diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/NetSocketWriter.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/NetSocketWriter.java index ac89512fdf..a695f354e8 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/NetSocketWriter.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/NetSocketWriter.java @@ -25,6 +25,7 @@ import java.nio.ByteBuffer; * data is wrapped in a wire protocol based on the messageFormat property. * * @author Gary Russell + * @since 2.0 */ public class NetSocketWriter extends AbstractSocketWriter { @@ -87,6 +88,8 @@ public class NetSocketWriter extends AbstractSocketWriter { protected void doClose() { try { socket.close(); - } catch (IOException e) {} + } catch (IOException e) { + logger.error("Error on close", e); + } } } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/NioSocketReader.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/NioSocketReader.java index a7baea6a40..c300d30a3e 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/NioSocketReader.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/NioSocketReader.java @@ -30,6 +30,7 @@ import org.springframework.integration.message.MessageMappingException; * A non-blocking SocketReader that reads from a {@link java.nio.channels.SocketChannel}. * * @author Gary Russell + * @since 2.0 * */ public class NioSocketReader extends AbstractSocketReader { @@ -71,13 +72,13 @@ public class NioSocketReader extends AbstractSocketReader { * @see org.springframework.integration.ip.tcp.SocketReader#assembleData() */ @Override - public boolean assembleDataLengthFormat() throws IOException { + public int assembleDataLengthFormat() throws IOException { if (lengthPart == null) { lengthPart = allocate(4); } if (lengthPart.hasRemaining()) { readChannel(lengthPart); - return false; + return MESSAGE_INCOMPLETE; } if (dataPart == null) { lengthPart.flip(); @@ -94,19 +95,19 @@ public class NioSocketReader extends AbstractSocketReader { if (dataPart.hasRemaining()) { readChannel(dataPart); if (dataPart.hasRemaining()) { - return false; + return MESSAGE_INCOMPLETE; } } assembledData = dataPart.array(); lengthPart = dataPart = null; - return true; + return MESSAGE_COMPLETE; } /* (non-Javadoc) * @see org.springframework.integration.ip.tcp.AbstractSocketReader#assembleDataStxEtxFormat() */ @Override - protected boolean assembleDataStxEtxFormat() throws IOException { + protected int assembleDataStxEtxFormat() throws IOException { if (readChannelNonDeterministic()) { byte bite = rawBuffer.get(); int count = 0; @@ -120,12 +121,12 @@ public class NioSocketReader extends AbstractSocketReader { if (logger.isDebugEnabled()) { logger.debug("Incomplete message, consumed 1 byte"); } - return false; + return MESSAGE_INCOMPLETE; } } else { if (bite == ETX) { finishAssembly(); - return true; + return MESSAGE_COMPLETE; } buildBuffer.put(bite); count++; @@ -139,7 +140,7 @@ public class NioSocketReader extends AbstractSocketReader { if (logger.isDebugEnabled()) { logger.debug("Incomplete message, consumed " + count + " bytes"); } - return false; + return MESSAGE_INCOMPLETE; } bite = rawBuffer.get(); if (bite == ETX) { @@ -156,13 +157,13 @@ public class NioSocketReader extends AbstractSocketReader { logger.debug("Consumed " + count + " bytes"); } finishAssembly(); - return true; + return MESSAGE_COMPLETE; } else { if (logger.isDebugEnabled()) { logger.debug("Incomplete message, consumed 0 bytes"); } } - return false; + return MESSAGE_INCOMPLETE; } /** @@ -180,7 +181,7 @@ public class NioSocketReader extends AbstractSocketReader { * @see org.springframework.integration.ip.tcp.AbstractSocketReader#assembleDataCrLfFormat() */ @Override - protected boolean assembleDataCrLfFormat() throws IOException { + protected int assembleDataCrLfFormat() throws IOException { if (readChannelNonDeterministic()) { int count = 0; while (true) { @@ -188,7 +189,7 @@ public class NioSocketReader extends AbstractSocketReader { if (logger.isDebugEnabled()) { logger.debug("Incomplete message, consumed " + count + " bytes"); } - return false; + return MESSAGE_INCOMPLETE; } byte bite = rawBuffer.get(); if (bite == '\n' && buildBuffer.position() > 0) { @@ -209,13 +210,13 @@ public class NioSocketReader extends AbstractSocketReader { logger.debug("Consumed " + count + " bytes"); } finishAssembly(); - return true; + return MESSAGE_COMPLETE; } else { if (logger.isDebugEnabled()) { logger.debug("Incomplete message, consumed 0 bytes"); } } - return false; + return MESSAGE_INCOMPLETE; } /** @@ -226,7 +227,7 @@ public class NioSocketReader extends AbstractSocketReader { * */ @Override - protected boolean assembleDataCustomFormat() throws IOException { + protected int assembleDataCustomFormat() throws IOException { throw new UnsupportedOperationException("Need to subclass for this format"); } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/NioSocketWriter.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/NioSocketWriter.java index da3da239e0..cffb764c2d 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/NioSocketWriter.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/NioSocketWriter.java @@ -27,6 +27,7 @@ import java.util.concurrent.LinkedBlockingQueue; * data is wrapped in a wire protocol based on the messageFormat property. * * @author Gary Russell + * @since 2.0 */ public class NioSocketWriter extends AbstractSocketWriter { @@ -250,7 +251,9 @@ public class NioSocketWriter extends AbstractSocketWriter { protected void doClose() { try { channel.close(); - } catch (IOException e) {} + } catch (IOException e) { + logger.error("Error on close", e); + } } } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/SimpleTcpNetInboundGateway.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/SimpleTcpNetInboundGateway.java index ffef2171a7..8a4221d93d 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/SimpleTcpNetInboundGateway.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/SimpleTcpNetInboundGateway.java @@ -15,6 +15,7 @@ */ package org.springframework.integration.ip.tcp; +import java.io.IOException; import java.net.Socket; import java.net.SocketException; @@ -34,6 +35,7 @@ import org.springframework.integration.message.MessageMappingException; * number of concurrent connections expected. * * @author Gary Russell + * @since 2.0 * */ public class SimpleTcpNetInboundGateway extends AbstractMessagingGateway { @@ -62,6 +64,8 @@ public class SimpleTcpNetInboundGateway extends AbstractMessagingGateway { protected Class customSocketWriterClass; + protected boolean close; + @Override protected void doStart() { super.doStart(); @@ -86,6 +90,7 @@ public class SimpleTcpNetInboundGateway extends AbstractMessagingGateway { this.delegate.setSoTimeout(soTimeout); this.delegate.setTaskScheduler(getTaskScheduler()); this.delegate.setCustomSocketReaderClassName(customSocketReaderClassName); + this.delegate.setClose(close); super.onInit(); } @@ -189,6 +194,17 @@ public class SimpleTcpNetInboundGateway extends AbstractMessagingGateway { } } + /** + * @param close the close to set + */ + public void setClose(boolean close) { + this.close = close; + } + + public boolean isListening() { + return delegate.isListening(); + } + private class WriteCapableTcpNetReceivingChannelAdapter extends TcpNetReceivingChannelAdapter { /** @@ -206,6 +222,13 @@ public class SimpleTcpNetInboundGateway extends AbstractMessagingGateway { customSocketWriterClass, socket); try { writer.write(this.mapper.fromMessage(message)); + if (close) { + try { + socket.close(); + } catch (IOException ioe) { + logger.error("Error on close", ioe); + } + } } catch (Exception e) { throw new MessageMappingException("Failed to map and send response", e); } 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 c2618a71dc..5350a11e14 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 @@ -50,6 +50,8 @@ public class SimpleTcpNetOutboundGateway extends protected NetSocketReader reader; + protected boolean close; + /** * Constructs a SimpleTcpNetOutboundGateway that sends data to the * specified host and port, and waits for a response. @@ -67,9 +69,9 @@ public class SimpleTcpNetOutboundGateway extends @Override protected synchronized Object handleRequestMessage(Message requestMessage) { this.handler.handleMessage(requestMessage); + Socket socket = this.handler.getSocket(); if (this.reader == null || - this.reader.getSocket() != handler.getSocket()) { - Socket socket = this.handler.getSocket(); + this.reader.getSocket() != socket) { // might have re-opened on error this.reader = SocketIoUtils.createNetReader(this.messageFormat, this.customSocketReaderClass, socket, this.maxMessageSize, this.soReceiveBufferSize); @@ -77,6 +79,10 @@ public class SimpleTcpNetOutboundGateway extends try { this.reader.assembleData(); // Net... always returns true byte[] bytes = this.reader.getAssembledData(); + if (close) { + logger.debug("Closing socket because close=true"); + this.handler.close(); + } return bytes; } catch (Exception e) { this.reader = null; @@ -207,4 +213,11 @@ public class SimpleTcpNetOutboundGateway extends this.setOutputChannel(replyChannel); } + /** + * @param close the close to set + */ + public void setClose(boolean close) { + this.close = close; + } + } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/SocketReader.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/SocketReader.java index 640f6787d9..c632d19b3b 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/SocketReader.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/SocketReader.java @@ -23,19 +23,26 @@ import java.net.Socket; * General interface for assembling message data from a TCP/IP Socket. * Implementations for {@link java.net.Socket} and {@link java.nio.channels.SocketChannel} * are provided. + * * @author Gary Russell + * @since 2.0 * */ public interface SocketReader { + public static int MESSAGE_INCOMPLETE = 0; + + public static int MESSAGE_COMPLETE = 1; + /** * Reads the data the socket and assembles * packets of data into a complete message, depending on the format of that * data. - * @return true when the message is assembled. + * @return MESSAGE_COMPLETE when message is assembled, otherwise MESSAGE_IMCOMPLETE, or + * < 0 if socket closed before any data for a message is received. * @throws IOException */ - public boolean assembleData() throws IOException; + public int assembleData() throws IOException; /** * Retrieves the assembled tcp data or null if the data is not diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpNetReceivingChannelAdapter.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpNetReceivingChannelAdapter.java index 15019c4b32..38932f6192 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpNetReceivingChannelAdapter.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpNetReceivingChannelAdapter.java @@ -30,6 +30,7 @@ import org.springframework.integration.ip.util.SocketIoUtils; * accordingly. * * @author Gary Russell + * @since 2.0 * */ public class TcpNetReceivingChannelAdapter extends @@ -58,6 +59,7 @@ public class TcpNetReceivingChannelAdapter extends try { serverSocket = ServerSocketFactory.getDefault() .createServerSocket(port, Math.abs(poolSize)); + listening = true; while (true) { final Socket socket = serverSocket.accept(); setSocketOptions(socket); @@ -72,6 +74,7 @@ public class TcpNetReceivingChannelAdapter extends serverSocket.close(); } catch (IOException e1) {} } + listening = false; serverSocket = null; if (active) { logger.error("Error on ServerSocket", e); @@ -94,8 +97,21 @@ public class TcpNetReceivingChannelAdapter extends this.soReceiveBufferSize); while (true) { try { - if (reader.assembleData()) { + int messageStatus = reader.assembleData(); + if (messageStatus < 0) { + return; + } + if (messageStatus == SocketReader.MESSAGE_COMPLETE) { processMessage(reader); + if (close) { + logger.debug("Closing socket because close=true"); + try { + reader.getSocket().close(); + } catch (IOException ioe) { + logger.error("Error on close", ioe); + } + break; + } } } catch (Exception e) { logger.error("processMessage failed", e); diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpNetSendingMessageHandler.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpNetSendingMessageHandler.java index 2d1a6685ef..3d67f0b1b5 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpNetSendingMessageHandler.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpNetSendingMessageHandler.java @@ -25,6 +25,7 @@ import org.springframework.integration.ip.util.SocketIoUtils; /** * TCP Sending Channel Adapter that that uses a {@link java.net.Socket}. * @author Gary Russell + * @since 2.0 * */ public class TcpNetSendingMessageHandler extends @@ -57,6 +58,7 @@ public class TcpNetSendingMessageHandler extends protected synchronized SocketWriter getWriter() { if (this.writer == null) { try { + logger.debug("Opening new socket connection"); this.socket = SocketFactory.getDefault().createSocket(this.host, this.port); this.setSocketAttributes(socket); NetSocketWriter writer = SocketIoUtils.createNetWriter(messageFormat, @@ -85,4 +87,12 @@ public class TcpNetSendingMessageHandler extends } } + /** + * Close the underlying socket and prepare to establish a new socket on + * the next write. + */ + protected void close() { + this.writer.doClose(); + this.writer = null; + } } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioReceivingChannelAdapter.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioReceivingChannelAdapter.java index c9c829906f..1e778a7eae 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioReceivingChannelAdapter.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioReceivingChannelAdapter.java @@ -38,6 +38,7 @@ import org.springframework.integration.ip.util.SocketIoUtils; * number of threads is controlled by the poolSize property. * * @author Gary Russell + * @since 2.0 * */ public class TcpNioReceivingChannelAdapter extends @@ -65,6 +66,7 @@ public class TcpNioReceivingChannelAdapter extends protected void server() { try { serverChannel = ServerSocketChannel.open(); + listening = true; serverChannel.configureBlocking(false); serverChannel.socket().bind(new InetSocketAddress(port), Math.abs(poolSize)); @@ -76,6 +78,7 @@ public class TcpNioReceivingChannelAdapter extends try { serverChannel.close(); } catch (IOException e1) { } + listening = false; serverChannel = null; if (active) { logger.error("Error on ServerSocketChannel", e); @@ -169,11 +172,23 @@ public class TcpNioReceivingChannelAdapter extends private void doRead(SelectionKey key) { NioSocketReader reader = (NioSocketReader) key.attachment(); try { - if (reader.assembleData()) { + int messageStatus = reader.assembleData(); + if (messageStatus < 0) { + return; + } + if (messageStatus == SocketReader.MESSAGE_COMPLETE) { Message message; message = mapper.toMessage(reader); if (message != null) { sendMessage(message); + if (close) { + logger.debug("Closing channel because close=true"); + try { + key.channel().close(); + } catch (IOException ioe) { + logger.error("Error on close", ioe); + } + } } } } catch (Exception e) {} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioSendingMessageHandler.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioSendingMessageHandler.java index 9fca8d0c48..1f174e16d1 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioSendingMessageHandler.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioSendingMessageHandler.java @@ -21,7 +21,10 @@ import org.springframework.integration.ip.util.SocketIoUtils; /** + * TCP Sending Channel Adapter that that uses a {@link java.nio.channels.SocketChannel}. + * * @author Gary Russell + * @since 2.0 * */ public class TcpNioSendingMessageHandler extends @@ -49,6 +52,7 @@ public class TcpNioSendingMessageHandler extends protected synchronized SocketWriter getWriter() { if (this.socketChannel == null) { try { + logger.debug("Creating new SocketChannel"); this.socketChannel = SocketChannel.open(this.destinationAddress); this.setSocketAttributes(socketChannel.socket()); NioSocketWriter writer = SocketIoUtils.createNioWriter(messageFormat, diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java index 0701a529f8..6a4e3d583b 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java @@ -102,6 +102,8 @@ public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolRece this.threadPoolTaskScheduler.initialize(); } + listening = true; + // Do as little as possible here so we can loop around and catch the next packet. // Just schedule the packet for processing. while (this.active) { @@ -121,6 +123,7 @@ public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolRece throw new MessagingException("failed to receive DatagramPacket", e); } } + listening = false; } protected void sendAck(Message message) { diff --git a/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.0.xsd b/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.0.xsd index 3db1baf2f3..a66713d9f5 100644 --- a/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.0.xsd +++ b/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.0.xsd @@ -39,6 +39,15 @@ the custom message format. See java docs for TcpNetReceivingChannelAdapter and T + + + +If true, the socket will be closed after a message is received and sent to the +outbound channel. If false, the socket will remain open ready to receive the +next message. + + + @@ -202,6 +211,18 @@ the custom message format. See java docs for TcpNetSendingChannelAdapter and Tcp + + + +If true, for an outbound gateway, the socket will be closed after a response +is received and sent to the +reply channel. If false, the socket will remain open and be used to send the +next message. If true, for an inbound gateway, the socket will be closed after +the response is sent. If false, the socket will remain open and be used to +receive the next message. + + + diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml b/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml index 84110f179b..53f8616564 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml @@ -74,6 +74,7 @@ so-timeout="32" using-direct-buffers="true" using-nio="true" + close="true" /> + + + + diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java index 0338998bbe..f67aa3cc60 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java @@ -95,12 +95,21 @@ public class ParserUnitTests { TcpNetSendingMessageHandler tcpOutNet; @Autowired + @Qualifier(value="simpleInGateway") SimpleTcpNetInboundGateway simpleTcpNetInboundGateway; @Autowired @Qualifier(value="org.springframework.integration.ip.tcp.SimpleTcpNetOutboundGateway#0") SimpleTcpNetOutboundGateway simpleTcpNetOutboundGateway; + @Autowired + @Qualifier(value="simpleInGatewayClose") + SimpleTcpNetInboundGateway simpleTcpNetInboundGatewayClose; + + @Autowired + @Qualifier(value="org.springframework.integration.ip.tcp.SimpleTcpNetOutboundGateway#1") + SimpleTcpNetOutboundGateway simpleTcpNetOutboundGatewayClose; + @Test public void testInUdp() { DirectFieldAccessor dfa = new DirectFieldAccessor(udpIn); @@ -136,6 +145,7 @@ public class ParserUnitTests { assertEquals(29, dfa.getPropertyValue("receiveBufferSize")); assertEquals(30, dfa.getPropertyValue("soReceiveBufferSize")); assertEquals(32, dfa.getPropertyValue("soTimeout")); + assertEquals(false, dfa.getPropertyValue("close")); } @Test @@ -150,6 +160,7 @@ public class ParserUnitTests { assertEquals(29, dfa.getPropertyValue("receiveBufferSize")); assertEquals(30, dfa.getPropertyValue("soReceiveBufferSize")); assertEquals(32, dfa.getPropertyValue("soTimeout")); + assertEquals(true, dfa.getPropertyValue("close")); } @Test @@ -163,6 +174,7 @@ public class ParserUnitTests { assertEquals(29, dfa.getPropertyValue("receiveBufferSize")); assertEquals(30, dfa.getPropertyValue("soReceiveBufferSize")); assertEquals(32, dfa.getPropertyValue("soTimeout")); + assertEquals(false, dfa.getPropertyValue("close")); } @Test @@ -217,7 +229,6 @@ public class ParserUnitTests { assertEquals(53, dfa.getPropertyValue("soSendBufferSize")); assertEquals(54, dfa.getPropertyValue("soTimeout")); assertEquals(false, dfa.getPropertyValue("usingDirectBuffers")); - } @Test @@ -233,7 +244,6 @@ public class ParserUnitTests { assertEquals(53, dfa.getPropertyValue("soSendBufferSize")); assertEquals(54, dfa.getPropertyValue("soTimeout")); assertEquals(true, dfa.getPropertyValue("usingDirectBuffers")); - } @Test @@ -248,7 +258,6 @@ public class ParserUnitTests { assertEquals(27, dfa.getPropertyValue("soTrafficClass")); assertEquals(53, dfa.getPropertyValue("soSendBufferSize")); assertEquals(54, dfa.getPropertyValue("soTimeout")); - } @Test @@ -267,13 +276,13 @@ public class ParserUnitTests { assertEquals(125, dfa.getPropertyValue("soSendBufferSize")); assertEquals(126, dfa.getPropertyValue("soTimeout")); assertEquals(23, dfa.getPropertyValue("poolSize")); - + assertEquals(false, dfa.getPropertyValue("close")); } @Test public void testOutGateway() { DirectFieldAccessor dfa = new DirectFieldAccessor(simpleTcpNetOutboundGateway); - assertTrue(simpleTcpNetOutboundGateway.getPort() >= 6500); + assertTrue(simpleTcpNetOutboundGateway.getPort() >= 6600); assertEquals(MessageFormats.FORMAT_CRLF, dfa.getPropertyValue("messageFormat")); TcpNetSendingMessageHandler handler = (TcpNetSendingMessageHandler) dfa .getPropertyValue("handler"); @@ -284,6 +293,42 @@ public class ParserUnitTests { assertEquals(224, dfa.getPropertyValue("soReceiveBufferSize")); assertEquals(225, delegateDfa.getPropertyValue("soSendBufferSize")); assertEquals(226, delegateDfa.getPropertyValue("soTimeout")); + assertEquals(false, dfa.getPropertyValue("close")); } + @Test + public void testInGatewayClose() { + DirectFieldAccessor dfa = new DirectFieldAccessor(simpleTcpNetInboundGatewayClose); + assertTrue(simpleTcpNetInboundGatewayClose.getPort() >= 6700); + assertEquals(MessageFormats.FORMAT_CRLF, dfa.getPropertyValue("messageFormat")); + TcpNetReceivingChannelAdapter delegate = (TcpNetReceivingChannelAdapter) dfa + .getPropertyValue("delegate"); + DirectFieldAccessor delegateDfa = new DirectFieldAccessor(delegate); + assertEquals(CustomNetSocketReader.class, delegateDfa.getPropertyValue("customSocketReaderClass")); + assertEquals(CustomNetSocketWriter.class, dfa.getPropertyValue("customSocketWriterClass")); + assertEquals(true, dfa.getPropertyValue("soKeepAlive")); + assertEquals(123, dfa.getPropertyValue("receiveBufferSize")); + assertEquals(124, dfa.getPropertyValue("soReceiveBufferSize")); + assertEquals(125, dfa.getPropertyValue("soSendBufferSize")); + assertEquals(126, dfa.getPropertyValue("soTimeout")); + assertEquals(23, dfa.getPropertyValue("poolSize")); + assertEquals(true, dfa.getPropertyValue("close")); + } + + @Test + public void testOutGatewayClose() { + DirectFieldAccessor dfa = new DirectFieldAccessor(simpleTcpNetOutboundGatewayClose); + assertTrue(simpleTcpNetOutboundGatewayClose.getPort() >= 6800); + assertEquals(MessageFormats.FORMAT_CRLF, dfa.getPropertyValue("messageFormat")); + TcpNetSendingMessageHandler handler = (TcpNetSendingMessageHandler) dfa + .getPropertyValue("handler"); + DirectFieldAccessor delegateDfa = new DirectFieldAccessor(handler); + assertEquals(CustomNetSocketReader.class, dfa.getPropertyValue("customSocketReaderClass")); + assertEquals(CustomNetSocketWriter.class, delegateDfa.getPropertyValue("customSocketWriterClass")); + assertEquals(true, delegateDfa.getPropertyValue("soKeepAlive")); + assertEquals(224, dfa.getPropertyValue("soReceiveBufferSize")); + assertEquals(225, delegateDfa.getPropertyValue("soSendBufferSize")); + assertEquals(226, delegateDfa.getPropertyValue("soTimeout")); + assertEquals(true, dfa.getPropertyValue("close")); + } } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/CustomNetSocketReader.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/CustomNetSocketReader.java index d2c160af20..6869ac62f5 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/CustomNetSocketReader.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/CustomNetSocketReader.java @@ -41,11 +41,14 @@ public class CustomNetSocketReader extends NetSocketReader { * @see org.springframework.integration.ip.tcp.NetSocketReader#assembleDataCustomFormat() */ @Override - protected boolean assembleDataCustomFormat() throws IOException { + protected int assembleDataCustomFormat() throws IOException { byte[] buff = new byte[24]; - read(buff); + int status = read(buff, true); + if (status < 0) { + return status; + } assembledData = buff; - return true; + return MESSAGE_COMPLETE; } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/CustomNioSocketReader.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/CustomNioSocketReader.java index 36d3efc88a..88081bb352 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/CustomNioSocketReader.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/CustomNioSocketReader.java @@ -44,17 +44,17 @@ public class CustomNioSocketReader extends NioSocketReader { * @see org.springframework.integration.ip.tcp.NetSocketReader#assembleDataCustomFormat() */ @Override - protected boolean assembleDataCustomFormat() throws IOException { + protected int assembleDataCustomFormat() throws IOException { if (buffer == null) { buffer = allocate(24); } readChannel(buffer); if (buffer.hasRemaining()) { - return false; + return MESSAGE_INCOMPLETE; } assembledData = buffer.array(); buffer = null; - return true; + return MESSAGE_COMPLETE; } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/NetSocketReaderTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/NetSocketReaderTests.java index 9181292695..6234b7dd0f 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/NetSocketReaderTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/NetSocketReaderTests.java @@ -45,14 +45,14 @@ public class NetSocketReaderTests { Socket socket = server.accept(); socket.setSoTimeout(5000); NetSocketReader reader = new NetSocketReader(socket); - if (reader.assembleData()) { + if (reader.assembleData() == SocketReader.MESSAGE_COMPLETE) { assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, new String(reader.getAssembledData())); } else { fail("Failed to assemble first message"); } - if (reader.assembleData()) { + if (reader.assembleData() == SocketReader.MESSAGE_COMPLETE) { assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, new String(reader.getAssembledData())); } @@ -75,14 +75,14 @@ public class NetSocketReaderTests { socket.setSoTimeout(5000); NetSocketReader reader = new NetSocketReader(socket); reader.setMessageFormat(MessageFormats.FORMAT_STX_ETX); - if (reader.assembleData()) { + if (reader.assembleData() == SocketReader.MESSAGE_COMPLETE) { assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, new String(reader.getAssembledData())); } else { fail("Failed to assemble first message"); } - if (reader.assembleData()) { + if (reader.assembleData() == SocketReader.MESSAGE_COMPLETE) { assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, new String(reader.getAssembledData())); } @@ -105,14 +105,14 @@ public class NetSocketReaderTests { socket.setSoTimeout(5000); NetSocketReader reader = new NetSocketReader(socket); reader.setMessageFormat(MessageFormats.FORMAT_CRLF); - if (reader.assembleData()) { + if (reader.assembleData() == SocketReader.MESSAGE_COMPLETE) { assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, new String(reader.getAssembledData())); } else { fail("Failed to assemble first message"); } - if (reader.assembleData()) { + if (reader.assembleData() == SocketReader.MESSAGE_COMPLETE) { assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, new String(reader.getAssembledData())); } @@ -131,7 +131,7 @@ public class NetSocketReaderTests { socket.setSoTimeout(5000); NetSocketReader reader = new NetSocketReader(socket); try { - if (reader.assembleData()) { + if (reader.assembleData() == SocketReader.MESSAGE_COMPLETE) { fail("Expected message length exceeded exception"); } } catch (IOException e) { @@ -153,7 +153,7 @@ public class NetSocketReaderTests { NetSocketReader reader = new NetSocketReader(socket); reader.setMessageFormat(MessageFormats.FORMAT_STX_ETX); try { - if (reader.assembleData()) { + if (reader.assembleData() == SocketReader.MESSAGE_COMPLETE) { fail("Expected message length exceeded exception"); } } catch (IOException e) { @@ -176,7 +176,7 @@ public class NetSocketReaderTests { reader.setMessageFormat(MessageFormats.FORMAT_STX_ETX); reader.setMaxMessageSize(1024); try { - if (reader.assembleData()) { + if (reader.assembleData() == SocketReader.MESSAGE_COMPLETE) { fail("Expected message length exceeded exception"); } } catch (IOException e) { @@ -198,7 +198,7 @@ public class NetSocketReaderTests { NetSocketReader reader = new NetSocketReader(socket); reader.setMessageFormat(MessageFormats.FORMAT_CRLF); try { - if (reader.assembleData()) { + if (reader.assembleData() == SocketReader.MESSAGE_COMPLETE) { fail("Expected message length exceeded exception"); } } catch (IOException e) { @@ -221,7 +221,7 @@ public class NetSocketReaderTests { reader.setMessageFormat(MessageFormats.FORMAT_CRLF); reader.setMaxMessageSize(1024); try { - if (reader.assembleData()) { + if (reader.assembleData() == SocketReader.MESSAGE_COMPLETE) { fail("Expected message length exceeded exception"); } } catch (IOException e) { diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/NioSocketReaderTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/NioSocketReaderTests.java index a0e72edbe0..c905dbc381 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/NioSocketReaderTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/NioSocketReaderTests.java @@ -82,7 +82,7 @@ public class NioSocketReaderTests { iterator.remove(); if (key.isReadable()) { assertEquals(channel, key.channel()); - if (reader.assembleData()) { + if (reader.assembleData() == SocketReader.MESSAGE_COMPLETE) { assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, new String(reader.getAssembledData())); count++; @@ -138,7 +138,7 @@ public class NioSocketReaderTests { iterator.remove(); if (key.isReadable()) { assertEquals(channel, key.channel()); - if (reader.assembleData()) { + if (reader.assembleData() == SocketReader.MESSAGE_COMPLETE) { assertEquals("Data", "xx", new String(reader.getAssembledData())); done = true; @@ -198,7 +198,7 @@ public class NioSocketReaderTests { iterator.remove(); if (key.isReadable()) { assertEquals(channel, key.channel()); - if (reader.assembleData()) { + if (reader.assembleData() == SocketReader.MESSAGE_COMPLETE) { assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, new String(reader.getAssembledData())); count++; @@ -258,7 +258,7 @@ public class NioSocketReaderTests { iterator.remove(); if (key.isReadable()) { assertEquals(channel, key.channel()); - if (reader.assembleData()) { + if (reader.assembleData() == SocketReader.MESSAGE_COMPLETE) { assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, new String(reader.getAssembledData())); count++; @@ -318,7 +318,7 @@ public class NioSocketReaderTests { if (key.isReadable()) { assertEquals(channel, key.channel()); try { - if (reader.assembleData()) { + if (reader.assembleData() == SocketReader.MESSAGE_COMPLETE) { fail("Expected message length exceeded exception"); } } catch (IOException e) { @@ -387,7 +387,7 @@ public class NioSocketReaderTests { if (key.isReadable()) { assertEquals(channel, key.channel()); try { - if (reader.assembleData()) { + if (reader.assembleData() == SocketReader.MESSAGE_COMPLETE) { fail("Expected message length exceeded exception"); } } catch (IOException e) { @@ -456,7 +456,7 @@ public class NioSocketReaderTests { if (key.isReadable()) { assertEquals(channel, key.channel()); try { - if (reader.assembleData()) { + if (reader.assembleData() == SocketReader.MESSAGE_COMPLETE) { fail("Expected message length exceeded exception"); } } catch (IOException e) { diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/SimpleTcpNetInboundGatewayTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/SimpleTcpNetInboundGatewayTests.java index 5f39dbdb61..4580661d75 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/SimpleTcpNetInboundGatewayTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/SimpleTcpNetInboundGatewayTests.java @@ -134,9 +134,9 @@ public class SimpleTcpNetInboundGatewayTests { startup = 0; Socket socket = SocketFactory.getDefault().createSocket("localhost", gatewayCustom.getPort()); String greetings = "Hello World!"; - String pad = " "; + String pad = " "; socket.getOutputStream().write((greetings).getBytes()); - socket.getOutputStream().write(pad.getBytes()); // will be truncated + socket.getOutputStream().write(pad.getBytes()); StringBuilder sb = new StringBuilder(); int c; int n = 0; 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 d1f668720e..ace42b4cc5 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 @@ -15,6 +15,8 @@ */ package org.springframework.integration.ip.tcp; +import static org.junit.Assert.assertEquals; + import java.net.ServerSocket; import java.net.Socket; import java.util.concurrent.CountDownLatch; @@ -22,7 +24,6 @@ import java.util.concurrent.TimeUnit; import javax.net.ServerSocketFactory; -import org.junit.Ignore; import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; @@ -35,8 +36,6 @@ import org.springframework.integration.message.MessageBuilder; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; -import static org.junit.Assert.assertEquals; - /** * @author Gary Russell * @@ -74,16 +73,29 @@ public class SimpleTcpNetOutboundGatewayTests { SimpleTcpNetOutboundGateway gateway = new SimpleTcpNetOutboundGateway ("localhost", inboundGatewayCrLf.getPort()); gateway.setMessageFormat(MessageFormats.FORMAT_CRLF); + waitListening(inboundGatewayCrLf); Message message = MessageBuilder.withPayload("test").build(); byte[] bytes = (byte[]) gateway.handleRequestMessage(message); assertEquals("echo:test", new String(bytes)); } + private void waitListening(SimpleTcpNetInboundGateway gateway) throws Exception { + int n = 0; + while (!gateway.isListening()) { + Thread.sleep(100); + if (n++ > 100) { + throw new Exception("Gateway failed to listen"); + } + } + + } + @Test public void testOutboundStxEtx() throws Exception { SimpleTcpNetOutboundGateway gateway = new SimpleTcpNetOutboundGateway ("localhost", inboundGatewayStxEtx.getPort()); gateway.setMessageFormat(MessageFormats.FORMAT_STX_ETX); + waitListening(inboundGatewayStxEtx); Message message = MessageBuilder.withPayload("test").build(); byte[] bytes = (byte[]) gateway.handleRequestMessage(message); assertEquals("echo:test", new String(bytes)); @@ -94,6 +106,7 @@ public class SimpleTcpNetOutboundGatewayTests { SimpleTcpNetOutboundGateway gateway = new SimpleTcpNetOutboundGateway ("localhost", inboundGatewayLength.getPort()); gateway.setMessageFormat(MessageFormats.FORMAT_LENGTH_HEADER); + waitListening(inboundGatewayLength); Message message = MessageBuilder.withPayload("test").build(); byte[] bytes = (byte[]) gateway.handleRequestMessage(message); assertEquals("echo:test", new String(bytes)); @@ -106,6 +119,7 @@ public class SimpleTcpNetOutboundGatewayTests { gateway.setMessageFormat(MessageFormats.FORMAT_CUSTOM); gateway.setCustomSocketReaderClassName("org.springframework.integration.ip.tcp.CustomNetSocketReader"); gateway.setCustomSocketWriterClassName("org.springframework.integration.ip.tcp.CustomNetSocketWriter"); + waitListening(inboundGatewayCustom); Message message = MessageBuilder.withPayload("test").build(); byte[] bytes = (byte[]) gateway.handleRequestMessage(message); assertEquals("echo:test", new String(bytes).trim()); @@ -119,11 +133,12 @@ public class SimpleTcpNetOutboundGatewayTests { assertEquals("echo:test", new String(bytes).trim()); } - @Ignore @Test + @Test public void testOutboundClose() 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 { @@ -134,8 +149,9 @@ public class SimpleTcpNetOutboundGatewayTests { byte[] b = new byte[1024]; s.getInputStream().read(b); s.getOutputStream().write("OK\r\n".getBytes()); - s.close(); latch2.countDown(); + latch3.await(); + s.close(); } } catch (Exception e) { e.printStackTrace(); @@ -146,9 +162,11 @@ public class SimpleTcpNetOutboundGatewayTests { SimpleTcpNetOutboundGateway gateway = new SimpleTcpNetOutboundGateway ("localhost", port); gateway.setMessageFormat(MessageFormats.FORMAT_CRLF); + gateway.setClose(true); Message message = MessageBuilder.withPayload("test").build(); byte[] bytes = (byte[]) gateway.handleRequestMessage(message); assertEquals("OK", new String(bytes)); + latch3.countDown(); latch2.await(2000, TimeUnit.MILLISECONDS); bytes = (byte[]) gateway.handleRequestMessage(message); assertEquals("OK", new String(bytes)); diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/SocketMessageMapperTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/SocketMessageMapperTests.java index c551e2bcc7..8d7ed761cf 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/SocketMessageMapperTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/SocketMessageMapperTests.java @@ -91,8 +91,8 @@ public class SocketMessageMapperTests { /* (non-Javadoc) * @see org.springframework.integration.ip.tcp.SocketReader#assembleData() */ - public boolean assembleData() { - return false; + public int assembleData() { + return SocketReader.MESSAGE_INCOMPLETE; } /* (non-Javadoc) 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 622f1dae6b..9b2a246266 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 @@ -18,6 +18,8 @@ package org.springframework.integration.ip.tcp; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; +import java.util.concurrent.CountDownLatch; + import org.junit.Test; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.core.Message; @@ -44,14 +46,13 @@ public class TcpReceivingChannelAdapterTests { taskScheduler.initialize(); adapter.setTaskScheduler(taskScheduler); adapter.start(); - Thread.sleep(2000); // wait for server to start listening + waitListening(adapter); SocketUtils.testSendLength(port, null); //sends 2 copies of TEST_STRING twice - Thread.sleep(2000); // wait for asynch processing - Message message = channel.receive(0); + Message message = channel.receive(2000); assertNotNull(message); assertEquals(SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, new String((byte[])message.getPayload())); - message = channel.receive(0); + message = channel.receive(2000); assertNotNull(message); assertEquals(SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, new String((byte[])message.getPayload())); @@ -74,14 +75,13 @@ public class TcpReceivingChannelAdapterTests { taskScheduler.initialize(); adapter.setTaskScheduler(taskScheduler); adapter.start(); - Thread.sleep(2000); // wait for server to start listening + waitListening(adapter); SocketUtils.testSendStxEtx(port, null); //sends 2 copies of TEST_STRING twice - Thread.sleep(4000); // wait for asynch processing - Message message = channel.receive(0); + Message message = channel.receive(4000); assertNotNull(message); assertEquals("\u0002" + SocketUtils.TEST_STRING + SocketUtils.TEST_STRING + "\u0003", new String((byte[])message.getPayload())); - message = channel.receive(0); + message = channel.receive(2000); assertNotNull(message); assertEquals("\u0002" + SocketUtils.TEST_STRING + SocketUtils.TEST_STRING + "\u0003", new String((byte[])message.getPayload())); @@ -102,14 +102,13 @@ public class TcpReceivingChannelAdapterTests { taskScheduler.initialize(); adapter.setTaskScheduler(taskScheduler); adapter.start(); - Thread.sleep(2000); // wait for server to start listening + waitListening(adapter); SocketUtils.testSendLength(port, null); //sends 2 copies of TEST_STRING twice - Thread.sleep(2000); // wait for asynch processing - Message message = channel.receive(0); + Message message = channel.receive(2000); assertNotNull(message); assertEquals(SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, new String((byte[])message.getPayload())); - message = channel.receive(0); + message = channel.receive(2000); assertNotNull(message); assertEquals(SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, new String((byte[])message.getPayload())); @@ -131,18 +130,98 @@ public class TcpReceivingChannelAdapterTests { taskScheduler.initialize(); adapter.setTaskScheduler(taskScheduler); adapter.start(); - Thread.sleep(2000); // wait for server to start listening + waitListening(adapter); SocketUtils.testSendStxEtx(port, null); //sends 2 copies of TEST_STRING twice - Thread.sleep(4000); // wait for asynch processing - Message message = channel.receive(0); + Message message = channel.receive(2000); assertNotNull(message); assertEquals("\u0002" + SocketUtils.TEST_STRING + SocketUtils.TEST_STRING + "\u0003", new String((byte[])message.getPayload())); - message = channel.receive(0); + message = channel.receive(2000); assertNotNull(message); assertEquals("\u0002" + SocketUtils.TEST_STRING + SocketUtils.TEST_STRING + "\u0003", new String((byte[])message.getPayload())); adapter.stop(); } + /** + * Tests close option on inbound adapter. + * + * @throws Exception + */ + @Test + public void testNetClose() throws Exception { + QueueChannel channel = new QueueChannel(2); + int port = SocketUtils.findAvailableServerSocket(); + AbstractTcpReceivingChannelAdapter adapter = new TcpNetReceivingChannelAdapter(port); + adapter.setOutputChannel(channel); + adapter.setClose(true); + adapter.setMessageFormat(MessageFormats.FORMAT_CRLF); + ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); + taskScheduler.initialize(); + adapter.setTaskScheduler(taskScheduler); + adapter.start(); + waitListening(adapter); + CountDownLatch latch = new CountDownLatch(1); + SocketUtils.testSendCrLfSingle(port, latch); + Message message = channel.receive(5000); + latch.countDown(); + assertNotNull(message); + assertEquals(SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, + new String((byte[])message.getPayload())); + latch = new CountDownLatch(1); + SocketUtils.testSendCrLfSingle(port, latch); + message = channel.receive(5000); + latch.countDown(); + assertNotNull(message); + assertEquals(SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, + new String((byte[])message.getPayload())); + adapter.stop(); + } + + /** + * Tests close option on inbound adapter. + * + * @throws Exception + */ + @Test + public void testNioClose() throws Exception { + QueueChannel channel = new QueueChannel(2); + int port = SocketUtils.findAvailableServerSocket(); + AbstractTcpReceivingChannelAdapter adapter = new TcpNioReceivingChannelAdapter(port); + adapter.setOutputChannel(channel); + adapter.setClose(true); + adapter.setMessageFormat(MessageFormats.FORMAT_CRLF); + ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); + taskScheduler.initialize(); + adapter.setTaskScheduler(taskScheduler); + adapter.start(); + waitListening(adapter); + CountDownLatch latch = new CountDownLatch(1); + SocketUtils.testSendCrLfSingle(port, latch); + Message message = channel.receive(2000); + assertNotNull(message); + assertEquals(SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, + new String((byte[])message.getPayload())); + latch = new CountDownLatch(1); + SocketUtils.testSendCrLfSingle(port, latch); + message = channel.receive(2000); + assertNotNull(message); + assertEquals(SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, + new String((byte[])message.getPayload())); + adapter.stop(); + } + + private void waitListening(AbstractInternetProtocolReceivingChannelAdapter adapter) throws Exception { + int n = 0; + while (!adapter.isListening()) { + Thread.sleep(100); + if (n++ > 100) { + throw new Exception("Gateway failed to listen"); + } + } + + } + + + } 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 c44b26fae0..841308e01e 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 @@ -222,7 +222,34 @@ public class SocketUtils { thread.setDaemon(true); thread.start(); } - + + /** + * Sends a single message +CRLF. + * @param latch Waits for latch to count down before closing the socket. + */ + public static void testSendCrLfSingle(final int port, final CountDownLatch latch) { + Thread thread = new Thread(new Runnable() { + public void run() { + try { + Socket socket = new Socket(InetAddress.getByName("localhost"), port); + OutputStream outputStream = socket.getOutputStream(); + outputStream.write(TEST_STRING.getBytes()); + outputStream.write(TEST_STRING.getBytes()); + writeByte(outputStream, '\r', true); + writeByte(outputStream, '\n', true); + if (latch != null) { + latch.await(); + } + socket.close(); + } catch (Exception e) { + e.printStackTrace(); + } + } + }); + thread.setDaemon(true); + thread.start(); + } + /** * Sends a large CRLF message with no CRLF. */