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 c5a901708e..c24b013b2d 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 @@ -223,6 +223,9 @@ public abstract class IpAdapterParserUtils { if (messageFormat.equals("crlf")) { return MessageFormats.FORMAT_CRLF; } + if (messageFormat.equals("serialized")) { + return MessageFormats.FORMAT_JAVA_SERIALIZED; + } if (messageFormat.equals("custom")) { return MessageFormats.FORMAT_CUSTOM; } 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 813b0780a9..643cc1c8b4 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 @@ -44,31 +44,42 @@ public abstract class AbstractSocketReader implements SocketReader, MessageForma * The assembled data; must contain a reference when assembleData() * returns true; will be set to null when getAssembledData() is called. */ - protected byte[] assembledData; + protected Object assembledData; protected int maxMessageSize = 1024 * 60; /** * Assembles data in format {@link #FORMAT_LENGTH_HEADER}. - * @return True when a message is completely assembled. + * @return SocketReader.MESSAGE_COMPLETE when message is assembled, otherwise SocketReader.MESSAGE_IMCOMPLETE, or + * < 0 if socket closed before any data for a message is received. * @throws IOException */ protected abstract int assembleDataLengthFormat() throws IOException; /** * Assembles data in format {@link #FORMAT_STX_ETX}. - * @return True when a message is completely assembled. + * @return SocketReader.MESSAGE_COMPLETE when message is assembled, otherwise SocketReader.MESSAGE_IMCOMPLETE, or + * < 0 if socket closed before any data for a message is received. * @throws IOException */ protected abstract int assembleDataStxEtxFormat() throws IOException; /** * Assembles data in format {@link #FORMAT_CRLF}. - * @return True when a message is completely assembled. + * @return SocketReader.MESSAGE_COMPLETE when message is assembled, otherwise SocketReader.MESSAGE_IMCOMPLETE, or + * < 0 if socket closed before any data for a message is received. * @throws IOException */ protected abstract int assembleDataCrLfFormat() throws IOException; - + + /** + * Assembles data in format {@link #FORMAT_JAVA_SERIALIZED} + * @return SocketReader.MESSAGE_COMPLETE when message is assembled, otherwise SocketReader.MESSAGE_IMCOMPLETE, or + * < 0 if socket closed before any data for a message is received. + * @throws IOException + */ + protected abstract int assembleDataSerializedFormat() throws IOException; + /** * Assembles data in format {@link #FORMAT_CUSTOM}. Implementations must * return false until the message is completely assembled, at which time @@ -90,6 +101,8 @@ public abstract class AbstractSocketReader implements SocketReader, MessageForma return assembleDataCrLfFormat(); case FORMAT_CUSTOM: return assembleDataCustomFormat(); + case FORMAT_JAVA_SERIALIZED: + return assembleDataSerializedFormat(); default: throw new UnsupportedOperationException( "Unsupported message format: " + messageFormat); @@ -100,6 +113,19 @@ public abstract class AbstractSocketReader implements SocketReader, MessageForma } } + /* (non-Javadoc) + * @see org.springframework.integration.ip.tcp.SocketReader#getAssembledData() + */ + public Object getAssembledData() { + Object assembledData = this.assembledData; + this.assembledData = null; + if (assembledData instanceof byte[] && + ((byte[]) assembledData).length == 0) { + return null; + } + return assembledData; + } + /** * Called after an exception; close the transport. */ 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 5ee036aa85..3277cf9576 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 @@ -48,20 +48,23 @@ public abstract class AbstractSocketWriter implements SocketWriter, MessageForma /* * @see org.springframework.integration.ip.tcp.SocketWriter#write(byte[]) */ - public synchronized void write(byte[] bytes) throws IOException { + public synchronized void write(Object object) throws IOException { try { switch (this.messageFormat) { case FORMAT_LENGTH_HEADER: - writeLengthFormat(bytes); + writeLengthFormat((byte[]) object); return; case FORMAT_STX_ETX: - writeStxEtxFormat(bytes); + writeStxEtxFormat((byte[]) object); return; case FORMAT_CRLF: - writeCrLfFormat(bytes); + writeCrLfFormat((byte[]) object); + return; + case FORMAT_JAVA_SERIALIZED: + writeSerializedFormat(object); return; case FORMAT_CUSTOM: - writeCustomFormat(bytes); + writeCustomFormat(object); return; default: throw new UnsupportedOperationException( @@ -101,11 +104,17 @@ public abstract class AbstractSocketWriter implements SocketWriter, MessageForma */ protected abstract void writeCrLfFormat(byte[] bytes) throws IOException; + /** + * Write the data, followed by carriage return, line feed ('\r\n'). + * @param bytes + */ + protected abstract void writeSerializedFormat(Object object) throws IOException; + /** * Write the data using some custom protocol. * @param bytes */ - protected abstract void writeCustomFormat(byte[] bytes) throws IOException; + protected abstract void writeCustomFormat(Object object) throws IOException; /** * @param messageFormat the messageFormat to set 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 02e68257b7..cfba314d1f 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 @@ -116,6 +116,7 @@ public abstract class AbstractTcpReceivingChannelAdapter extends */ public void setMessageFormat(int messageFormat) { this.messageFormat = messageFormat; + mapper.setMessageFormat(messageFormat); } /** diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/AbstractTcpSendingMessageHandler.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/AbstractTcpSendingMessageHandler.java index d7a5dddbcb..75ad38d000 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/AbstractTcpSendingMessageHandler.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/AbstractTcpSendingMessageHandler.java @@ -114,12 +114,12 @@ public abstract class AbstractTcpSendingMessageHandler extends */ protected void doWrite(Message message) { try { - byte[] bytes = mapper.fromMessage(message); + Object object = mapper.fromMessage(message); SocketWriter writer = this.getWriter(); if (writer == null) { throw new MessageMappingException(message, "Failed to create SocketWriter"); } - writer.write(bytes); + writer.write(object); } catch (Exception e) { this.writer = null; if (e instanceof MessageMappingException) { @@ -168,6 +168,7 @@ public abstract class AbstractTcpSendingMessageHandler extends */ public void setMessageFormat(int messageFormat) { this.messageFormat = messageFormat; + mapper.setMessageFormat(messageFormat); } } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/MessageFormats.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/MessageFormats.java index c64b09b565..c386b0e864 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/MessageFormats.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/MessageFormats.java @@ -35,10 +35,20 @@ public interface MessageFormats { * Message has format '<message>\r\n'. */ public static final int FORMAT_CRLF = 3; + /** + * Not a real format; any formats less than this must use + * a byte[] payload (or a String for outbound). Formats higher + * have implied format (e.g. Serializable). + */ + public static final int FORMAT_IMPLICIT = 128; + /** + * Message contains a Serializable object + */ + public static final int FORMAT_JAVA_SERIALIZED = 129; /** * Message has custom format. */ - public static final int FORMAT_CUSTOM = 99; + public static final int FORMAT_CUSTOM = 255; public static final int STX = 0x02; 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 5810d2e5f3..97614ff96d 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 @@ -15,8 +15,10 @@ */ package org.springframework.integration.ip.tcp; +import java.io.EOFException; import java.io.IOException; import java.io.InputStream; +import java.io.ObjectInputStream; import java.net.InetAddress; import java.net.Socket; import java.nio.ByteBuffer; @@ -40,6 +42,8 @@ public class NetSocketReader extends AbstractSocketReader { protected final Log logger = LogFactory.getLog(getClass()); protected Socket socket; + + protected ObjectInputStream objectInputStream; /** * Constructs a NetsocketReader which reads from the Socket. @@ -68,7 +72,7 @@ public class NetSocketReader extends AbstractSocketReader { } byte[] messagePart = new byte[messageLength]; read(messagePart, false); - assembledData = messagePart; + this.assembledData = messagePart; return MESSAGE_COMPLETE; } @@ -94,8 +98,8 @@ public class NetSocketReader extends AbstractSocketReader { + this.maxMessageSize); } } - assembledData = new byte[n]; - System.arraycopy(buffer, 0, assembledData, 0, n); + this.assembledData = new byte[n]; + System.arraycopy(buffer, 0, this.assembledData, 0, n); return MESSAGE_COMPLETE; } @@ -129,11 +133,27 @@ public class NetSocketReader extends AbstractSocketReader { + this.maxMessageSize); } }; - assembledData = new byte[n-1]; - System.arraycopy(buffer, 0, assembledData, 0, n-1); + this.assembledData = new byte[n-1]; + System.arraycopy(buffer, 0, this.assembledData, 0, n-1); return MESSAGE_COMPLETE; } + @Override + protected int assembleDataSerializedFormat() throws IOException { + try { + if (this.objectInputStream == null) { + InputStream is = this.socket.getInputStream(); + this.objectInputStream = new ObjectInputStream(is); + } + this.assembledData = this.objectInputStream.readObject(); + } catch (EOFException ee) { + return -1; + } catch (ClassNotFoundException e) { + throw new IOException(e); + } + return SocketReader.MESSAGE_COMPLETE; + } + /** * Throws {@link UnsupportedOperationException}; custom implementations can * subclass this class and provide an implementation for this method. @@ -146,15 +166,6 @@ public class NetSocketReader extends AbstractSocketReader { throw new UnsupportedOperationException("Need to subclass for this format"); } - /* (non-Javadoc) - * @see org.springframework.integration.ip.tcp.SocketReader#getAssembledData() - */ - public byte[] getAssembledData() { - byte[] assembledData = this.assembledData; - this.assembledData = null; - return assembledData; - } - /** * Reads data from the socket and puts the data in buffer. Blocks until * buffer is full or a socket timeout occurs. @@ -168,7 +179,7 @@ public class NetSocketReader extends AbstractSocketReader { int needed = buffer.length; while (lengthRead < needed) { int len; - len = socket.getInputStream().read(buffer, lengthRead, + len = this.socket.getInputStream().read(buffer, lengthRead, needed - lengthRead); if (len < 0 && header && lengthRead == 0) { return len; 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 a695f354e8..a7006048f1 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 @@ -16,6 +16,7 @@ package org.springframework.integration.ip.tcp; import java.io.IOException; +import java.io.ObjectOutputStream; import java.io.OutputStream; import java.net.Socket; import java.nio.ByteBuffer; @@ -30,6 +31,8 @@ import java.nio.ByteBuffer; public class NetSocketWriter extends AbstractSocketWriter { protected Socket socket; + + protected ObjectOutputStream objectOutputStream; /** * Constructs a NetSocketWriter for the Socket. @@ -45,17 +48,31 @@ public class NetSocketWriter extends AbstractSocketWriter { */ @Override protected void writeCrLfFormat(byte[] bytes) throws IOException { - OutputStream outputStream = socket.getOutputStream(); + OutputStream outputStream = this.socket.getOutputStream(); outputStream.write(bytes); outputStream.write('\r'); outputStream.write('\n'); + outputStream.flush(); } /* (non-Javadoc) * @see org.springframework.integration.ip.tcp.AbstractSocketWriter#writeCustomFormat(byte[]) */ @Override - protected void writeCustomFormat(byte[] bytes) throws IOException { + protected void writeSerializedFormat(Object object) throws IOException { + if (this.objectOutputStream == null) { + OutputStream os = this.socket.getOutputStream(); + this.objectOutputStream = new ObjectOutputStream(os); + } + this.objectOutputStream.writeObject(object); + this.objectOutputStream.flush(); + } + + /* (non-Javadoc) + * @see org.springframework.integration.ip.tcp.AbstractSocketWriter#writeCustomFormat(byte[]) + */ + @Override + protected void writeCustomFormat(Object object) throws IOException { throw new UnsupportedOperationException("Need to subclass for this format"); } @@ -66,19 +83,21 @@ public class NetSocketWriter extends AbstractSocketWriter { protected void writeLengthFormat(byte[] bytes) throws IOException { ByteBuffer lengthPart = ByteBuffer.allocate(4); lengthPart.putInt(bytes.length); - OutputStream outputStream = socket.getOutputStream(); + OutputStream outputStream = this.socket.getOutputStream(); outputStream.write(lengthPart.array()); outputStream.write(bytes); + outputStream.flush(); } /* (non-Javadoc) * @see org.springframework.integration.ip.tcp.AbstractSocketWriter#writeStxEtxFormat(byte[]) */ @Override protected void writeStxEtxFormat(byte[] bytes) throws IOException { - OutputStream outputStream = socket.getOutputStream(); + OutputStream outputStream = this.socket.getOutputStream(); outputStream.write(STX); outputStream.write(bytes); outputStream.write(ETX); + outputStream.flush(); } /* (non-Javadoc) @@ -87,7 +106,7 @@ public class NetSocketWriter extends AbstractSocketWriter { @Override protected void doClose() { try { - socket.close(); + this.socket.close(); } 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 c300d30a3e..34e6d37467 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 @@ -59,15 +59,6 @@ public class NioSocketReader extends AbstractSocketReader { this.channel = channel; } - /* (non-Javadoc) - * @see org.springframework.integration.ip.tcp.SocketReader#read(java.nio.ByteBuffer, int) - */ - public byte[] getAssembledData() { - byte[] assembledData = this.assembledData; - this.assembledData = null; - return assembledData; - } - /* (non-Javadoc) * @see org.springframework.integration.ip.tcp.SocketReader#assembleData() */ @@ -170,10 +161,11 @@ public class NioSocketReader extends AbstractSocketReader { * */ private void finishAssembly() { - assembledData = new byte[buildBuffer.position()]; + byte[] assembledData = new byte[buildBuffer.position()]; System.arraycopy(buildBuffer.array(), 0, assembledData, 0, assembledData.length); building = false; buildBuffer.clear(); + this.assembledData = assembledData; logger.debug("Message assembly complete"); } @@ -219,6 +211,18 @@ public class NioSocketReader extends AbstractSocketReader { return MESSAGE_INCOMPLETE; } + /** + * Throws {@link UnsupportedOperationException}; Java serialization is currently only + * supported using the NetSocketReader. + * @throws IOException + * @see org.springframework.integration.ip.tcp.AbstractSocketReader#assembleDataCustomFormat(). + * + */ + protected int assembleDataSerializedFormat() throws IOException { + throw new UnsupportedOperationException("Serializable not supported using NIO"); + } + + /** * Throws {@link UnsupportedOperationException}; custom implementations can * subclass this class and provide an implementation. 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 cffb764c2d..21020c5850 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 @@ -104,7 +104,7 @@ public class NioSocketWriter extends AbstractSocketWriter { return buffer; } synchronized (buffers) { - if (bufferCount < maxBuffers) { + if (this.bufferCount < this.maxBuffers) { bufferCount++; return ByteBuffer.allocateDirect(this.sendBufferSize); } @@ -116,7 +116,7 @@ public class NioSocketWriter extends AbstractSocketWriter { protected void returnBuffer(ByteBuffer buffer) { if (buffer != null) { - buffers.offer(buffer); + this.buffers.offer(buffer); } } @@ -126,7 +126,7 @@ public class NioSocketWriter extends AbstractSocketWriter { @Override protected void writeCrLfFormat(byte[] bytes) throws IOException { ByteBuffer buffer = null; - if (usingDirectBuffers) { + if (this.usingDirectBuffers) { try { checkBufferSize(bytes, 2); buffer = getBuffer(); @@ -134,7 +134,7 @@ public class NioSocketWriter extends AbstractSocketWriter { buffer.put((byte) '\r'); buffer.put((byte) '\n'); buffer.flip(); - channel.write(buffer); + this.channel.write(buffer); return; } catch (InterruptedException e) { throw new IOException("Could not get buffer; interrupted"); @@ -143,14 +143,14 @@ public class NioSocketWriter extends AbstractSocketWriter { } } synchronized (channel) { - if (crLfPart == null) { - crLfPart = ByteBuffer.allocate(2); - crLfPart.put((byte) '\r'); - crLfPart.put((byte) '\n'); + if (this.crLfPart == null) { + this.crLfPart = ByteBuffer.allocate(2); + this.crLfPart.put((byte) '\r'); + this.crLfPart.put((byte) '\n'); } - channel.write(ByteBuffer.wrap(bytes)); - crLfPart.flip(); - channel.write(crLfPart); + this.channel.write(ByteBuffer.wrap(bytes)); + this.crLfPart.flip(); + this.channel.write(this.crLfPart); } } @@ -158,7 +158,16 @@ public class NioSocketWriter extends AbstractSocketWriter { * @see org.springframework.integration.ip.tcp.AbstractSocketWriter#writeCustomFormat(byte[]) */ @Override - protected void writeCustomFormat(byte[] bytes) throws IOException { + protected void writeSerializedFormat(Object object) throws IOException { + throw new UnsupportedOperationException("Serializable not supported using NIO"); + } + + /* (non-Javadoc) + * @see org.springframework.integration.ip.tcp.AbstractSocketWriter#writeCustomFormat(byte[]) + */ + @Override + protected void writeCustomFormat(Object object) throws IOException { + throw new UnsupportedOperationException("Need to subclass for this format"); } /* (non-Javadoc) @@ -167,14 +176,14 @@ public class NioSocketWriter extends AbstractSocketWriter { @Override protected void writeLengthFormat(byte[] bytes) throws IOException { ByteBuffer buffer = null; - if (usingDirectBuffers) { + if (this.usingDirectBuffers) { try { checkBufferSize(bytes, 4); buffer = getBuffer(); buffer.putInt(bytes.length); buffer.put(bytes); buffer.flip(); - channel.write(buffer); + this.channel.write(buffer); return; } catch (InterruptedException e) { throw new IOException("Could not get buffer; interrupted"); @@ -183,15 +192,15 @@ public class NioSocketWriter extends AbstractSocketWriter { } } synchronized (channel) { - if (lengthPart == null) { - lengthPart = ByteBuffer.allocate(4); + if (this.lengthPart == null) { + this.lengthPart = ByteBuffer.allocate(4); } else { - lengthPart.clear(); + this.lengthPart.clear(); } - lengthPart.putInt(bytes.length); - lengthPart.flip(); - channel.write(lengthPart); - channel.write(ByteBuffer.wrap(bytes)); + this.lengthPart.putInt(bytes.length); + this.lengthPart.flip(); + this.channel.write(this.lengthPart); + this.channel.write(ByteBuffer.wrap(bytes)); } } @@ -201,7 +210,7 @@ public class NioSocketWriter extends AbstractSocketWriter { @Override protected void writeStxEtxFormat(byte[] bytes) throws IOException { ByteBuffer buffer = null; - if (usingDirectBuffers) { + if (this.usingDirectBuffers) { try { checkBufferSize(bytes, 2); buffer = getBuffer(); @@ -209,7 +218,7 @@ public class NioSocketWriter extends AbstractSocketWriter { buffer.put(bytes); buffer.put((byte) ETX); buffer.flip(); - channel.write(buffer); + this.channel.write(buffer); return; } catch (InterruptedException e) { throw new IOException("Could not get buffer; interrupted"); @@ -218,17 +227,17 @@ public class NioSocketWriter extends AbstractSocketWriter { } } synchronized (channel) { - if (stxPart == null) { - stxPart = ByteBuffer.allocate(1); - stxPart.put((byte) STX); - etxPart = ByteBuffer.allocate(1); - etxPart.put((byte) ETX); + if (this.stxPart == null) { + this.stxPart = ByteBuffer.allocate(1); + this.stxPart.put((byte) STX); + this.etxPart = ByteBuffer.allocate(1); + this.etxPart.put((byte) ETX); } - stxPart.flip(); - channel.write(stxPart); - channel.write(ByteBuffer.wrap(bytes)); - etxPart.flip(); - channel.write(etxPart); + this.stxPart.flip(); + this.channel.write(this.stxPart); + this.channel.write(ByteBuffer.wrap(bytes)); + this.etxPart.flip(); + this.channel.write(this.etxPart); } } @@ -237,7 +246,7 @@ public class NioSocketWriter extends AbstractSocketWriter { * @throws IOException */ private void checkBufferSize(byte[] bytes, int pad) throws IOException { - if (bytes.length + pad > sendBufferSize) { + if (bytes.length + pad > this.sendBufferSize) { throw new IOException("Send buffer too small (" + sendBufferSize + ") increase so-send-buffer-size to at least " + bytes.length + pad); @@ -250,7 +259,7 @@ public class NioSocketWriter extends AbstractSocketWriter { @Override protected void doClose() { try { - channel.close(); + this.channel.close(); } 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 8a4221d93d..7def8795aa 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 @@ -120,6 +120,7 @@ public class SimpleTcpNetInboundGateway extends AbstractMessagingGateway { */ public void setMessageFormat(int messageFormat) { this.messageFormat = messageFormat; + mapper.setMessageFormat(messageFormat); } /** 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 5350a11e14..d64ada4960 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 @@ -78,12 +78,12 @@ public class SimpleTcpNetOutboundGateway extends } try { this.reader.assembleData(); // Net... always returns true - byte[] bytes = this.reader.getAssembledData(); + Object object = this.reader.getAssembledData(); if (close) { logger.debug("Closing socket because close=true"); this.handler.close(); } - return bytes; + return object; } catch (Exception e) { this.reader = null; throw new MessagingException(requestMessage, e); diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/SocketMessageMapper.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/SocketMessageMapper.java index 1a2335ee53..450900b19d 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/SocketMessageMapper.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/SocketMessageMapper.java @@ -33,14 +33,16 @@ import org.springframework.integration.message.OutboundMessageMapper; */ public class SocketMessageMapper implements InboundMessageMapper, - OutboundMessageMapper { + OutboundMessageMapper { private volatile String charset = "UTF-8"; + private volatile int messageFormat; + /* (non-Javadoc) * @see org.springframework.integration.message.InboundMessageMapper#toMessage(java.lang.Object) */ - public Message toMessage(SocketReader socketReader) throws Exception { + public Message toMessage(SocketReader socketReader) throws Exception { return fromRaw(socketReader); } @@ -53,10 +55,10 @@ public class SocketMessageMapper implements * @return * @throws IOException */ - private Message fromRaw(SocketReader socketReader) throws IOException { - byte[] payload = socketReader.getAssembledData(); - Message message = null; - if (payload != null && payload.length > 0) { + private Message fromRaw(SocketReader socketReader) throws IOException { + Object payload = socketReader.getAssembledData(); + Message message = null; + if (payload != null) { message = MessageBuilder.withPayload(payload) .setHeader(IpHeaders.HOSTNAME, socketReader.getAddress().getHostName()) .setHeader(IpHeaders.IP_ADDRESS, socketReader.getAddress().getHostAddress()) @@ -69,8 +71,11 @@ public class SocketMessageMapper implements /* (non-Javadoc) * @see org.springframework.integration.message.OutboundMessageMapper#fromMessage(org.springframework.integration.core.Message) */ - public byte[] fromMessage(Message message) throws Exception { - return getPayloadAsBytes(message); + public Object fromMessage(Message message) throws Exception { + if (this.messageFormat < MessageFormats.FORMAT_IMPLICIT) { + return getPayloadAsBytes(message); + } + return message.getPayload(); } /** @@ -107,4 +112,9 @@ public class SocketMessageMapper implements this.charset = charset; } + + public void setMessageFormat(int messageFormat) { + this.messageFormat = messageFormat; + } + } 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 c632d19b3b..827e603cf9 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 @@ -50,7 +50,7 @@ public interface SocketReader { * again null until a new assembly is completed. * @return The assembled data or null. */ - public byte[] getAssembledData(); + public Object getAssembledData(); /** * Returns the InetAddress of the underlying socket. diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/SocketWriter.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/SocketWriter.java index cada12e902..8168979b95 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/SocketWriter.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/SocketWriter.java @@ -33,7 +33,7 @@ public interface SocketWriter { * @param bytes The bytes to write. * @throws IOException */ - void write(byte[] bytes) throws IOException; + void write(Object object) throws IOException; /** * @param messageFormat the messageFormat to set 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 78cea2dff5..c33cc6ae76 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 @@ -124,7 +124,7 @@ public class TcpNetReceivingChannelAdapter extends protected void processMessage(NetSocketReader reader) throws Exception { - Message message = mapper.toMessage(reader); + Message message = mapper.toMessage(reader); if (message != null) { sendMessage(message); } 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 5cabc88d78..f6d59e2344 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 @@ -185,7 +185,7 @@ public class TcpNioReceivingChannelAdapter extends logger.error("Error on close", ioe); } } - Message message; + Message message; message = mapper.toMessage(reader); if (message != null) { sendMessage(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 a66713d9f5..de89190af5 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 @@ -239,6 +239,7 @@ 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 53f8616564..5385d938ae 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 @@ -91,6 +91,19 @@ so-timeout="32" /> + + + + = 5450); + assertEquals(MessageFormats.FORMAT_JAVA_SERIALIZED, dfa.getPropertyValue("messageFormat")); + assertEquals(27, dfa.getPropertyValue("poolSize")); + assertEquals(true, dfa.getPropertyValue("soKeepAlive")); + assertEquals(29, dfa.getPropertyValue("receiveBufferSize")); + assertEquals(30, dfa.getPropertyValue("soReceiveBufferSize")); + assertEquals(32, dfa.getPropertyValue("soTimeout")); + assertEquals(false, dfa.getPropertyValue("close")); + } + @Test public void testOutUdp() { DirectFieldAccessor dfa = new DirectFieldAccessor(udpOut); @@ -260,6 +281,19 @@ public class ParserUnitTests { assertEquals(54, dfa.getPropertyValue("soTimeout")); } + @Test + public void testOutTcpNetSerialized() { + DirectFieldAccessor dfa = new DirectFieldAccessor(tcpOutNetSerialized); + assertTrue(tcpOutNetSerialized.getPort() >= 6450); + assertEquals(MessageFormats.FORMAT_JAVA_SERIALIZED, dfa.getPropertyValue("messageFormat")); + assertEquals(true, dfa.getPropertyValue("soKeepAlive")); + assertEquals(3, dfa.getPropertyValue("soLinger")); + assertEquals(true, dfa.getPropertyValue("soTcpNoDelay")); + assertEquals(27, dfa.getPropertyValue("soTrafficClass")); + assertEquals(53, dfa.getPropertyValue("soSendBufferSize")); + assertEquals(54, dfa.getPropertyValue("soTimeout")); + } + @Test public void testInGateway() { DirectFieldAccessor dfa = new DirectFieldAccessor(simpleTcpNetInboundGateway); diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/CustomNetSocketWriter.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/CustomNetSocketWriter.java index 5269b618a6..651e6148c2 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/CustomNetSocketWriter.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/CustomNetSocketWriter.java @@ -36,7 +36,16 @@ public class CustomNetSocketWriter extends NetSocketWriter { * @see org.springframework.integration.ip.tcp.NetSocketWriter#writeCustomFormat(byte[]) */ @Override - protected void writeCustomFormat(byte[] bytes) throws IOException { + protected void writeCustomFormat(Object object) throws IOException { + byte[] bytes; + if (object instanceof byte[]) { + bytes = (byte[]) object; + } else if (object instanceof String) { + bytes = ((String) object).getBytes(); + } else { + throw new UnsupportedOperationException("Only supports String and byte[]"); + } + if (bytes.length > 24) { socket.getOutputStream().write(bytes, 0, 24); return; diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/CustomNioSocketWriter.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/CustomNioSocketWriter.java index 3baf8f3ab9..ab7cb21964 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/CustomNioSocketWriter.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/CustomNioSocketWriter.java @@ -41,7 +41,16 @@ public class CustomNioSocketWriter extends NioSocketWriter { * @see org.springframework.integration.ip.tcp.NetSocketWriter#writeCustomFormat(byte[]) */ @Override - protected void writeCustomFormat(byte[] bytes) throws IOException { + protected void writeCustomFormat(Object object) throws IOException { + byte[] bytes; + if (object instanceof byte[]) { + bytes = (byte[]) object; + } else if (object instanceof String) { + bytes = ((String) object).getBytes(); + } else { + throw new UnsupportedOperationException("Only supports String and byte[]"); + } + ByteBuffer data = ByteBuffer.wrap(bytes); if (bytes.length > 24) { data.limit(24); 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 6234b7dd0f..18d106a65b 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 @@ -47,14 +47,14 @@ public class NetSocketReaderTests { NetSocketReader reader = new NetSocketReader(socket); if (reader.assembleData() == SocketReader.MESSAGE_COMPLETE) { assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, - new String(reader.getAssembledData())); + new String((byte[]) reader.getAssembledData())); } else { fail("Failed to assemble first message"); } if (reader.assembleData() == SocketReader.MESSAGE_COMPLETE) { assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, - new String(reader.getAssembledData())); + new String((byte[]) reader.getAssembledData())); } else { fail("Failed to assemble second message"); @@ -77,14 +77,14 @@ public class NetSocketReaderTests { reader.setMessageFormat(MessageFormats.FORMAT_STX_ETX); if (reader.assembleData() == SocketReader.MESSAGE_COMPLETE) { assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, - new String(reader.getAssembledData())); + new String((byte[]) reader.getAssembledData())); } else { fail("Failed to assemble first message"); } if (reader.assembleData() == SocketReader.MESSAGE_COMPLETE) { assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, - new String(reader.getAssembledData())); + new String((byte[]) reader.getAssembledData())); } else { fail("Failed to assemble second message"); @@ -107,14 +107,44 @@ public class NetSocketReaderTests { reader.setMessageFormat(MessageFormats.FORMAT_CRLF); if (reader.assembleData() == SocketReader.MESSAGE_COMPLETE) { assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, - new String(reader.getAssembledData())); + new String((byte[]) reader.getAssembledData())); } else { fail("Failed to assemble first message"); } if (reader.assembleData() == SocketReader.MESSAGE_COMPLETE) { assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, - new String(reader.getAssembledData())); + new String((byte[]) reader.getAssembledData())); + } + else { + fail("Failed to assemble second message"); + } + server.close(); + } + + /** + * Test method for {@link org.springframework.integration.ip.tcp.NioSocketReader#readFully()}, + * using STX<message>ETX + */ + @Test + public void testReadSerialized() throws Exception { + int port = SocketUtils.findAvailableServerSocket(); + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + SocketUtils.testSendSerialized(port); + Socket socket = server.accept(); + socket.setSoTimeout(5000); + NetSocketReader reader = new NetSocketReader(socket); + reader.setMessageFormat(MessageFormats.FORMAT_JAVA_SERIALIZED); + if (reader.assembleData() == SocketReader.MESSAGE_COMPLETE) { + assertEquals("Data", SocketUtils.TEST_STRING, + reader.getAssembledData()); + } + else { + fail("Failed to assemble first message"); + } + if (reader.assembleData() == SocketReader.MESSAGE_COMPLETE) { + assertEquals("Data", SocketUtils.TEST_STRING, + reader.getAssembledData()); } else { fail("Failed to assemble second message"); diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/NetSocketWriterTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/NetSocketWriterTests.java index 4faea4561f..7107ea21d4 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/NetSocketWriterTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/NetSocketWriterTests.java @@ -19,6 +19,7 @@ import static org.junit.Assert.*; import java.io.IOException; import java.io.InputStream; +import java.io.ObjectInputStream; import java.net.ServerSocket; import java.net.Socket; import java.nio.ByteBuffer; @@ -131,6 +132,35 @@ public class NetSocketWriterTests { server.close(); } + @Test + public void testWriteSerialized() throws Exception { + final int port = SocketUtils.findAvailableServerSocket(); + final String testString = "abcdef"; + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + Thread t = new Thread(new Runnable() { + public void run() { + try { + Socket socket = SocketFactory.getDefault().createSocket("localhost", port); + NetSocketWriter writer = new NetSocketWriter(socket); + writer.setMessageFormat(MessageFormats.FORMAT_JAVA_SERIALIZED); + writer.write(testString); + writer.write(testString); + Thread.sleep(1000000000L); + } catch (Exception e) { + e.printStackTrace(); + } + } + }); + t.setDaemon(true); + t.start(); + Socket socket = server.accept(); + InputStream is = socket.getInputStream(); + ObjectInputStream ois = new ObjectInputStream(is); + assertEquals(testString, ois.readObject()); + assertEquals(testString, ois.readObject()); + server.close(); + } + /** * @param is * @param buff 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 c905dbc381..d9304f55df 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 @@ -84,7 +84,7 @@ public class NioSocketReaderTests { assertEquals(channel, key.channel()); if (reader.assembleData() == SocketReader.MESSAGE_COMPLETE) { assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, - new String(reader.getAssembledData())); + new String((byte[]) reader.getAssembledData())); count++; } latch.countDown(); @@ -140,7 +140,7 @@ public class NioSocketReaderTests { assertEquals(channel, key.channel()); if (reader.assembleData() == SocketReader.MESSAGE_COMPLETE) { assertEquals("Data", "xx", - new String(reader.getAssembledData())); + new String((byte[]) reader.getAssembledData())); done = true; } latch.countDown(); @@ -200,7 +200,7 @@ public class NioSocketReaderTests { assertEquals(channel, key.channel()); if (reader.assembleData() == SocketReader.MESSAGE_COMPLETE) { assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, - new String(reader.getAssembledData())); + new String((byte[]) reader.getAssembledData())); count++; } latch.countDown(); @@ -260,7 +260,7 @@ public class NioSocketReaderTests { assertEquals(channel, key.channel()); if (reader.assembleData() == SocketReader.MESSAGE_COMPLETE) { assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, - new String(reader.getAssembledData())); + new String((byte[]) reader.getAssembledData())); count++; } latch.countDown(); diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/SimpleTcpNetInboundGatewayTests-context.xml b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/SimpleTcpNetInboundGatewayTests-context.xml index 4af4c674e6..9f54137ac2 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/SimpleTcpNetInboundGatewayTests-context.xml +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/SimpleTcpNetInboundGatewayTests-context.xml @@ -27,6 +27,11 @@ request-channel="toSA" message-format="length-header" /> + + 100) { + throw new Exception("Gateway failed to listen"); + } + } + + } + } 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 ace42b4cc5..fde11a4abb 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 @@ -56,6 +56,10 @@ public class SimpleTcpNetOutboundGatewayTests { @Qualifier("gatewayLength") private SimpleTcpNetInboundGateway inboundGatewayLength; + @Autowired + @Qualifier("gatewaySerialized") + private SimpleTcpNetInboundGateway inboundGatewaySerialized; + @Autowired @Qualifier("gatewayCustom") private SimpleTcpNetInboundGateway inboundGatewayCustom; @@ -101,6 +105,17 @@ public class SimpleTcpNetOutboundGatewayTests { assertEquals("echo:test", new String(bytes)); } + @Test + public void testOutboundSerialized() throws Exception { + SimpleTcpNetOutboundGateway gateway = new SimpleTcpNetOutboundGateway + ("localhost", inboundGatewaySerialized.getPort()); + gateway.setMessageFormat(MessageFormats.FORMAT_JAVA_SERIALIZED); + waitListening(inboundGatewaySerialized); + Message message = MessageBuilder.withPayload("test").build(); + Object response = gateway.handleRequestMessage(message); + assertEquals("echo:test", response); + } + @Test public void testOutboundLength() throws Exception { SimpleTcpNetOutboundGateway gateway = new SimpleTcpNetOutboundGateway 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 8d7ed761cf..40daeea8c2 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 @@ -45,8 +45,8 @@ public class SocketMessageMapperTests { @Test public void testToMessage() throws Exception { SocketMessageMapper mapper = new SocketMessageMapper(); - Message message = mapper.toMessage(new StubSocketReader()); - assertEquals(TEST_PAYLOAD, new String(message.getPayload())); + Message message = mapper.toMessage(new StubSocketReader()); + assertEquals(TEST_PAYLOAD, new String((byte[]) message.getPayload())); assertEquals(InetAddress.getLocalHost().getHostName(), message .getHeaders().get(IpHeaders.HOSTNAME)); assertEquals(InetAddress.getLocalHost().getHostAddress(), message @@ -64,7 +64,7 @@ public class SocketMessageMapperTests { String s = "test"; Message message = MessageBuilder.withPayload(s).build(); SocketMessageMapper mapper = new SocketMessageMapper(); - byte[] bArray = mapper.fromMessage(message); + byte[] bArray = (byte[]) mapper.fromMessage(message); assertEquals(s, new String(bArray)); } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TestService.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TestService.java index fa6514ff77..516e24759b 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TestService.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TestService.java @@ -26,4 +26,9 @@ public class TestService { public String test(byte[] bytes) { return "echo:" + new String(bytes); } + + public String test(String s) { + return "echo:" + s; + } + } 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 841308e01e..2dc18e8e0b 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 @@ -15,6 +15,7 @@ */ package org.springframework.integration.ip.util; +import java.io.ObjectOutputStream; import java.io.OutputStream; import java.net.DatagramSocket; import java.net.InetAddress; @@ -250,6 +251,31 @@ public class SocketUtils { thread.start(); } + /** + * Sends two serialized objects over the same socket. + * @param port + */ + public static void testSendSerialized(final int port) { + Thread thread = new Thread(new Runnable() { + public void run() { + try { + Socket socket = new Socket(InetAddress.getByName("localhost"), port); + OutputStream outputStream = socket.getOutputStream(); + ObjectOutputStream oos = new ObjectOutputStream(outputStream); + oos.writeObject(TEST_STRING); + oos.flush(); + oos.writeObject(TEST_STRING); + oos.flush(); + Thread.sleep(1000000000L); // wait forever, but we're a daemon + } catch (Exception e) { + e.printStackTrace(); + } + } + }); + thread.setDaemon(true); + thread.start(); + } + /** * Sends a large CRLF message with no CRLF. */