Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-1150 Support java serialization over TCPNet* adapters and Simple TCP gateways.

Browse Source
This commit is contained in:
Gary Russell
2010-05-29 16:56:50 +00:00
parent 3b7696449b
commit 4fa1483e42
31 changed files with 435 additions and 116 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java
View File

@@ -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;
}

36
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/AbstractSocketReader.java
View File

@@ -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.
*/

21
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/AbstractSocketWriter.java
View File

@@ -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

1
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/AbstractTcpReceivingChannelAdapter.java
View File

@@ -116,6 +116,7 @@ public abstract class AbstractTcpReceivingChannelAdapter extends
*/
public void setMessageFormat(int messageFormat) {
this.messageFormat = messageFormat;
mapper.setMessageFormat(messageFormat);
}
/**

5
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/AbstractTcpSendingMessageHandler.java
View File

@@ -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);
}
}

12
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/MessageFormats.java
View File

@@ -35,10 +35,20 @@ public interface MessageFormats {
* Message has format '&lt;message&gt;\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;

41
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/NetSocketReader.java
View File

@@ -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;

29
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/NetSocketWriter.java
View File

@@ -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);
}

24
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/NioSocketReader.java
View File

@@ -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.

79
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/NioSocketWriter.java
View File

@@ -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);
}

1
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/SimpleTcpNetInboundGateway.java
View File

@@ -120,6 +120,7 @@ public class SimpleTcpNetInboundGateway extends AbstractMessagingGateway {
*/
public void setMessageFormat(int messageFormat) {
this.messageFormat = messageFormat;
mapper.setMessageFormat(messageFormat);
}
/**

4
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/SimpleTcpNetOutboundGateway.java
View File

@@ -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);

26
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/SocketMessageMapper.java
View File

@@ -33,14 +33,16 @@ import org.springframework.integration.message.OutboundMessageMapper;
*/
public class SocketMessageMapper implements
InboundMessageMapper<SocketReader>,
OutboundMessageMapper<byte[]> {
OutboundMessageMapper<Object> {
private volatile String charset = "UTF-8";
private volatile int messageFormat;
/* (non-Javadoc)
* @see org.springframework.integration.message.InboundMessageMapper#toMessage(java.lang.Object)
*/
public Message<byte[]> toMessage(SocketReader socketReader) throws Exception {
public Message<Object> toMessage(SocketReader socketReader) throws Exception {
return fromRaw(socketReader);
}
@@ -53,10 +55,10 @@ public class SocketMessageMapper implements
* @return
* @throws IOException
*/
private Message<byte[]> fromRaw(SocketReader socketReader) throws IOException {
byte[] payload = socketReader.getAssembledData();
Message<byte[]> message = null;
if (payload != null && payload.length > 0) {
private Message<Object> fromRaw(SocketReader socketReader) throws IOException {
Object payload = socketReader.getAssembledData();
Message<Object> 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;
}
}

2
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/SocketReader.java
View File

@@ -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.

2
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/SocketWriter.java
View File

@@ -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

2
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpNetReceivingChannelAdapter.java
View File

@@ -124,7 +124,7 @@ public class TcpNetReceivingChannelAdapter extends
protected void processMessage(NetSocketReader reader)
throws Exception {
Message<byte[]> message = mapper.toMessage(reader);
Message<Object> message = mapper.toMessage(reader);
if (message != null) {
sendMessage(message);
}

2
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioReceivingChannelAdapter.java
View File

@@ -185,7 +185,7 @@ public class TcpNioReceivingChannelAdapter extends
logger.error("Error on close", ioe);
}
}
Message<byte[]> message;
Message<Object> message;
message = mapper.toMessage(reader);
if (message != null) {
sendMessage(message);

1
spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.0.xsd
View File

@@ -239,6 +239,7 @@ receive the next message.
<xsd:enumeration value="length-header" />
<xsd:enumeration value="stx-etx" />
<xsd:enumeration value="crlf" />
<xsd:enumeration value="serialized" />
<xsd:enumeration value="custom" />
</xsd:restriction>
</xsd:simpleType>

27
spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml
View File

@@ -91,6 +91,19 @@
so-timeout="32"
/>
<ip:inbound-channel-adapter id="testInTcpNetSerialized"
channel="tcpChannel"
check-length="true"
message-format="serialized"
pool-size="27"
port="#{tcpIpUtils.findAvailableServerSocket(5450)}"
protocol="tcp"
receive-buffer-size="29"
so-keep-alive="true"
so-receive-buffer-size="30"
so-timeout="32"
/>
<ip:outbound-channel-adapter id="testOutUdp"
ack-host="somehost"
ack-port="#{tcpIpUtils.findAvailableUdpSocket(7000)}"
@@ -175,6 +188,20 @@
so-traffic-class="27"
/>
<ip:outbound-channel-adapter id="testOutTcpNetSerialized"
channel="tcpChannel"
host="localhost"
port="#{tcpIpUtils.findAvailableServerSocket(6450)}"
protocol="tcp"
so-send-buffer-size="53"
so-timeout="54"
message-format="serialized"
so-keep-alive="true"
so-linger="3"
so-tcp-no-delay="true"
so-traffic-class="27"
/>
<ip:inbound-gateway id="simpleInGateway"
request-channel="tcpChannel"
reply-channel="replyChannel"

34
spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java
View File

@@ -74,6 +74,10 @@ public class ParserUnitTests {
@Qualifier(value="testInTcpNet")
TcpNetReceivingChannelAdapter tcpInNet;
@Autowired
@Qualifier(value="testInTcpNetSerialized")
TcpNetReceivingChannelAdapter tcpInNetSerialized;
@Autowired
@Qualifier(value="org.springframework.integration.ip.udp.UnicastSendingMessageHandler#0")
UnicastSendingMessageHandler udpOut;
@@ -94,6 +98,10 @@ public class ParserUnitTests {
@Qualifier(value="org.springframework.integration.ip.tcp.TcpNetSendingMessageHandler#0")
TcpNetSendingMessageHandler tcpOutNet;
@Autowired
@Qualifier(value="org.springframework.integration.ip.tcp.TcpNetSendingMessageHandler#1")
TcpNetSendingMessageHandler tcpOutNetSerialized;
@Autowired
@Qualifier(value="simpleInGateway")
SimpleTcpNetInboundGateway simpleTcpNetInboundGateway;
@@ -177,6 +185,19 @@ public class ParserUnitTests {
assertEquals(false, dfa.getPropertyValue("close"));
}
@Test
public void testInTcpNetSerialized() {
DirectFieldAccessor dfa = new DirectFieldAccessor(tcpInNetSerialized);
assertTrue(tcpInNetSerialized.getPort() >= 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);

11
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/CustomNetSocketWriter.java
View File

@@ -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;

11
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/CustomNioSocketWriter.java
View File

@@ -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);

42
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/NetSocketReaderTests.java
View File

@@ -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&lt;message&gt;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");

30
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/NetSocketWriterTests.java
View File

@@ -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

8
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/NioSocketReaderTests.java
View File

@@ -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();

5
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/SimpleTcpNetInboundGatewayTests-context.xml
View File

@@ -27,6 +27,11 @@
request-channel="toSA"
message-format="length-header" />
<ip:inbound-gateway id="gatewaySerialized"
port="#{tcpIpUtils.findAvailableServerSocket(5450)}"
request-channel="toSA"
message-format="serialized" />
<ip:inbound-gateway id="gatewayCustom"
port="#{tcpIpUtils.findAvailableServerSocket(5500)}"
request-channel="toSA"

40
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/SimpleTcpNetInboundGatewayTests.java
View File

@@ -17,6 +17,8 @@ package org.springframework.integration.ip.tcp;
import static org.junit.Assert.assertEquals;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.net.Socket;
import javax.net.SocketFactory;
@@ -50,14 +52,17 @@ public class SimpleTcpNetInboundGatewayTests {
@Qualifier(value="gatewayLength")
SimpleTcpNetInboundGateway gatewayLength;
@Autowired
@Qualifier(value="gatewaySerialized")
SimpleTcpNetInboundGateway gatewaySerialized;
@Autowired
@Qualifier(value="gatewayCustom")
SimpleTcpNetInboundGateway gatewayCustom;
@Test
public void testCrLf() throws Exception {
Thread.sleep(startup);
startup = 0;
waitListening(gatewayCrLf);
Socket socket = SocketFactory.getDefault().createSocket("localhost", gatewayCrLf.getPort());
String greetings = "Hello World!";
socket.getOutputStream().write((greetings + "\r\n").getBytes());
@@ -75,8 +80,7 @@ public class SimpleTcpNetInboundGatewayTests {
@Test
public void testStxEtx() throws Exception {
Thread.sleep(startup);
startup = 0;
waitListening(gatewayStxEtx);
Socket socket = SocketFactory.getDefault().createSocket("localhost", gatewayStxEtx.getPort());
String greetings = "Hello World!";
socket.getOutputStream().write(MessageFormats.STX);
@@ -97,10 +101,19 @@ public class SimpleTcpNetInboundGatewayTests {
assertEquals("echo:" + greetings, sb.toString());
}
@Test
public void testSerialized() throws Exception {
waitListening(gatewaySerialized);
Socket socket = SocketFactory.getDefault().createSocket("localhost", gatewaySerialized.getPort());
String greetings = "Hello World!";
new ObjectOutputStream(socket.getOutputStream()).writeObject(greetings);
String echo = (String) new ObjectInputStream(socket.getInputStream()).readObject();
assertEquals("echo:" + greetings, echo);
}
@Test
public void testLength() throws Exception {
Thread.sleep(startup);
startup = 0;
waitListening(gatewayLength);
Socket socket = SocketFactory.getDefault().createSocket("localhost", gatewayLength.getPort());
String greetings = "Hello World!";
byte[] header = new byte[4];
@@ -130,8 +143,7 @@ public class SimpleTcpNetInboundGatewayTests {
@Test
public void testCustom() throws Exception {
Thread.sleep(startup);
startup = 0;
waitListening(gatewayCustom);
Socket socket = SocketFactory.getDefault().createSocket("localhost", gatewayCustom.getPort());
String greetings = "Hello World!";
String pad = " ";
@@ -150,4 +162,16 @@ public class SimpleTcpNetInboundGatewayTests {
}
assertEquals("echo:" + greetings, sb.toString().trim());
}
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");
}
}
}
}

15
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/SimpleTcpNetOutboundGatewayTests.java
View File

@@ -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<String> message = MessageBuilder.withPayload("test").build();
Object response = gateway.handleRequestMessage(message);
assertEquals("echo:test", response);
}
@Test
public void testOutboundLength() throws Exception {
SimpleTcpNetOutboundGateway gateway = new SimpleTcpNetOutboundGateway

6
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/SocketMessageMapperTests.java
View File

@@ -45,8 +45,8 @@ public class SocketMessageMapperTests {
@Test
public void testToMessage() throws Exception {
SocketMessageMapper mapper = new SocketMessageMapper();
Message<byte[]> message = mapper.toMessage(new StubSocketReader());
assertEquals(TEST_PAYLOAD, new String(message.getPayload()));
Message<Object> 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<String> 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));
}

5
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TestService.java
View File

@@ -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;
}
}

26
spring-integration-ip/src/test/java/org/springframework/integration/ip/util/SocketUtils.java
View File

@@ -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.
*/