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

Merge branch 'master' of git.springsource.org:spring-integration/spring-integration

Browse Source
This commit is contained in:
Oleg Zhurakousky
2010-09-29 08:53:39 -04:00
parent 68d819feeb 3c93433baf
commit 2a6620b6e8
24 changed files with 235 additions and 227 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -45,7 +45,7 @@ public abstract class AbstractClientConnectionFactory extends AbstractConnection
}
/**
* Transfers attributes such as converters, singleUse etc to a new connection.
* Transfers attributes such as (de)serializers, singleUse etc to a new connection.
* When the connection factory has a reference to a TCPListener (to read
* responses), or for single use connections, the connection is executed.
* Single use connections need to read from the connection in order to

6
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java
View File

@@ -27,7 +27,7 @@ import org.apache.commons.logging.LogFactory;
import org.springframework.commons.serializer.Deserializer;
import org.springframework.commons.serializer.Serializer;
import org.springframework.context.SmartLifecycle;
import org.springframework.integration.ip.tcp.converter.ByteArrayCrLfConverter;
import org.springframework.integration.ip.tcp.serializer.ByteArrayCrLfSerializer;
import org.springframework.util.Assert;
/**
@@ -68,9 +68,9 @@ public abstract class AbstractConnectionFactory
protected Executor taskExecutor;
protected Deserializer<?> deserializer = new ByteArrayCrLfConverter();
protected Deserializer<?> deserializer = new ByteArrayCrLfSerializer();
protected Serializer<?> serializer = new ByteArrayCrLfConverter();
protected Serializer<?> serializer = new ByteArrayCrLfSerializer();
protected TcpMessageMapper mapper = new TcpMessageMapper();

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

@@ -60,7 +60,7 @@ public abstract class AbstractServerConnectionFactory extends AbstractConnection
}
/**
* Transfers attributes such as converters, singleUse etc to a new connection.
* Transfers attributes such as (de)serializer, singleUse etc to a new connection.
* For single use sockets, enforces a socket timeout (default 10 seconds).
* @param connection The new connection.
* @param socket The new socket.

6
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnection.java
View File

@@ -23,7 +23,7 @@ import org.apache.commons.logging.LogFactory;
import org.springframework.commons.serializer.Deserializer;
import org.springframework.commons.serializer.Serializer;
import org.springframework.integration.ip.tcp.converter.AbstractByteArrayStreamingConverter;
import org.springframework.integration.ip.tcp.serializer.AbstractByteArraySerializer;
import org.springframework.util.Assert;
/**
@@ -86,7 +86,7 @@ public abstract class AbstractTcpConnection implements TcpConnection {
Assert.notNull(mapper, this.getClass().getName() + " Mapper may not be null");
this.mapper = mapper;
if (this.serializer != null &&
!(this.serializer instanceof AbstractByteArrayStreamingConverter)) {
!(this.serializer instanceof AbstractByteArraySerializer)) {
mapper.setStringToBytes(false);
}
}
@@ -119,7 +119,7 @@ public abstract class AbstractTcpConnection implements TcpConnection {
*/
public void setSerializer(Serializer<?> serializer) {
this.serializer = serializer;
if (!(serializer instanceof AbstractByteArrayStreamingConverter)) {
if (!(serializer instanceof AbstractByteArraySerializer)) {
this.mapper.setStringToBytes(false);
}
}

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

@@ -25,7 +25,7 @@ import org.springframework.integration.Message;
/**
* An abstraction over {@link Socket} and {@link SocketChannel} that
* sends {@link Message} objects by converting the payload
* sends {@link Message} objects by serializing the payload
* and streaming it to the destination. Requires a {@link TcpListener}
* to receive incoming messages.
*
@@ -53,7 +53,7 @@ public interface TcpConnection extends Runnable {
public void send(Message<?> message) throws Exception;
/**
* Uses the input converter to obtain the message payload
* Uses the deserializer to obtain the message payload
* from the connection's input stream.
* @return The payload
* @throws Exception

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

@@ -88,7 +88,7 @@ public class TcpMessageMapper implements
}
else {
throw new MessageHandlingException(message,
"When using a byte array streaming converter, the socket mapper expects " +
"When using a byte array serializer, the socket mapper expects " +
"either a byte array or String payload, but received: " + payload.getClass());
}
return bytes;

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

@@ -19,9 +19,10 @@ package org.springframework.integration.ip.tcp.connection;
import java.net.Socket;
import java.net.SocketTimeoutException;
import org.springframework.commons.serializer.Deserializer;
import org.springframework.integration.Message;
import org.springframework.integration.ip.tcp.SocketIoUtils;
import org.springframework.integration.ip.tcp.converter.SoftEndOfStreamException;
import org.springframework.integration.ip.tcp.serializer.SoftEndOfStreamException;
/**
* A TcpConnection that uses and underlying {@link Socket}.
@@ -88,7 +89,7 @@ public class TcpNetConnection extends AbstractTcpConnection {
* If there is no listener, and this connection is not for single use,
* this method exits. When there is a listener, the method runs in a
* loop reading input from the connections's stream, data is converted
* to an object using the {@link InputStreamingConverter} and the listener's
* to an object using the {@link Deserializer} and the listener's
* {@link TcpListener#onMessage(Message)} method is called. For single use
* connections with no listener, the socket is closed after its timeout
* expires. If data is received on a single use socket with no listener,

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

@@ -31,7 +31,7 @@ import java.util.concurrent.atomic.AtomicInteger;
import org.springframework.integration.Message;
import org.springframework.integration.ip.tcp.SocketIoUtils;
import org.springframework.integration.ip.tcp.converter.SoftEndOfStreamException;
import org.springframework.integration.ip.tcp.serializer.SoftEndOfStreamException;
/**
* A TcpConnection that uses and underlying {@link SocketChannel}.

6
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/package-info.java
View File

@@ -1,6 +0,0 @@
/**
* Byte array converters for putting some protocol on the
* wire so we can delimit incoming messages.
*/
package org.springframework.integration.ip.tcp.converter;

11
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/AbstractByteArrayStreamingConverter.java → spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/AbstractByteArraySerializer.java
View File

@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.integration.ip.tcp.converter;
package org.springframework.integration.ip.tcp.serializer;
import java.io.IOException;
@@ -25,13 +25,14 @@ import org.springframework.commons.serializer.Deserializer;
import org.springframework.commons.serializer.Serializer;
/**
* Base class for streaming converters that convert to/from a byte array.
* Base class for (de)serializers that provide a mechanism to
* reconstruct a byte array from an arbitrary stream.
*
* @author Gary Russell
* @since 2.0
*
*/
public abstract class AbstractByteArrayStreamingConverter implements
public abstract class AbstractByteArraySerializer implements
Serializer<byte[]>,
Deserializer<byte[]> {
@@ -40,7 +41,7 @@ public abstract class AbstractByteArrayStreamingConverter implements
protected Log logger = LogFactory.getLog(this.getClass());
/**
* The maximum supported message size for this converter.
* The maximum supported message size for this serializer.
* Default 2048.
* @return The max message size.
*/
@@ -49,7 +50,7 @@ public abstract class AbstractByteArrayStreamingConverter implements
}
/**
* The maximum supported message size for this converter.
* The maximum supported message size for this serializer.
* Default 2048.
* @param maxMessageSize The max message size.
*/

8
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/ByteArrayCrLfConverter.java → spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayCrLfSerializer.java
View File

@@ -14,24 +14,24 @@
* limitations under the License.
*/
package org.springframework.integration.ip.tcp.converter;
package org.springframework.integration.ip.tcp.serializer;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
/**
* Converts data in an InputStream to a byte[]; data is terminated by \r\n
* Reads data in an InputStream to a byte[]; data must be terminated by \r\n
* (not included in resulting byte[]).
* Writes a byte[] to an OutputStream and adds \r\n.
*
* @author Gary Russell
* @since 2.0
*/
public class ByteArrayCrLfConverter extends AbstractByteArrayStreamingConverter {
public class ByteArrayCrLfSerializer extends AbstractByteArraySerializer {
/**
* Converts the data in the inputstream to a byte[]. Data must be terminated
* Reads the data in the inputstream to a byte[]. Data must be terminated
* by CRLF (\r\n). Throws a {@link SoftEndOfStreamException} if the stream
* is closed immediately after the \r\n (i.e. no data is in the process of
* being read).

8
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/ByteArrayLengthHeaderConverter.java → spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayLengthHeaderSerializer.java
View File

@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.integration.ip.tcp.converter;
package org.springframework.integration.ip.tcp.serializer;
import java.io.IOException;
import java.io.InputStream;
@@ -25,10 +25,10 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
/**
* Converts data in an InputStream to a byte[]; data is preceded by
* Reads data in an InputStream to a byte[]; data must be preceded by
* a 4 byte binary length (network byte order,
* not included in resulting byte[]).
* Writes a byte[] to an OutputStream after a 4 byte binary length.\
* Writes a byte[] to an OutputStream after a 4 byte binary length.
* The length field contains the length of data following the length
* field.
* (network byte order).
@@ -36,7 +36,7 @@ import org.apache.commons.logging.LogFactory;
* @author Gary Russell
* @since 2.0
*/
public class ByteArrayLengthHeaderConverter extends AbstractByteArrayStreamingConverter {
public class ByteArrayLengthHeaderSerializer extends AbstractByteArraySerializer {
private Log logger = LogFactory.getLog(this.getClass());

10
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/ByteArrayStxEtxConverter.java → spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayStxEtxSerializer.java
View File

@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.integration.ip.tcp.converter;
package org.springframework.integration.ip.tcp.serializer;
import java.io.IOException;
import java.io.InputStream;
@@ -23,21 +23,21 @@ import java.io.OutputStream;
import org.springframework.integration.mapping.MessageMappingException;
/**
* Converts data in an InputStream to a byte[]; data is prefixed by &lt;stx&gt; and
* Reads data in an InputStream to a byte[]; data must be prefixed by &lt;stx&gt; and
* terminated by &lt;etx&gt; (not included in resulting byte[]).
* Writes a byte[] to an OutputStream and prefixed by &lt;stx&gt; terminated by &lt;etx&gt;
* Writes a byte[] to an OutputStream prefixed by &lt;stx&gt; terminated by &lt;etx&gt;
*
* @author Gary Russell
* @since 2.0
*/
public class ByteArrayStxEtxConverter extends AbstractByteArrayStreamingConverter {
public class ByteArrayStxEtxSerializer extends AbstractByteArraySerializer {
public static final int STX = 0x02;
public static final int ETX = 0x03;
/**
* Converts the data in the inputstream to a byte[]. Data must be prefixed
* Reads the data in the inputstream to a byte[]. Data must be prefixed
* with an ASCII STX character, and terminated with an ASCII ETX character.
* Throws a {@link SoftEndOfStreamException} if the stream
* is closed immediately before the STX (i.e. no data is in the process of

2
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/SoftEndOfStreamException.java → spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/SoftEndOfStreamException.java
View File

@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.integration.ip.tcp.converter;
package org.springframework.integration.ip.tcp.serializer;
import java.io.IOException;

6
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/package-info.java Normal file
View File

@@ -0,0 +1,6 @@
/**
* Byte array (de)serializers for putting some protocol on the
* wire so that incoming messages can be constructed from stream data.
*/
package org.springframework.integration.ip.tcp.serializer;

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

@@ -299,7 +299,7 @@ the factory, the connection will be closed after a response is received.
</xsd:appinfo>
<xsd:documentation>
A Serializer that converts message payloads to/from output streams/input streams
associated with the connection. Default is ByteArrayCrLfConverter. Serializer and Deserializer
associated with the connection. Default is ByteArrayCrLfSerializer. Serializer and Deserializer
would normally be the same but this is not required.
</xsd:documentation>
</xsd:annotation>
@@ -313,7 +313,7 @@ would normally be the same but this is not required.
</xsd:appinfo>
<xsd:documentation>
A Deserializer that converts message payloads to/from output streams/input streams
associated with the connection. Default is ByteArrayCrLfConverter. Serializer and Deserializer
associated with the connection. Default is ByteArrayCrLfSerializer. Serializer and Deserializer
would normally be the same but this is not required.
</xsd:documentation>
</xsd:annotation>

10
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpConfigInboundGatewayTests.java
View File

@@ -34,7 +34,7 @@ import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.support.AbstractApplicationContext;
import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory;
import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory;
import org.springframework.integration.ip.tcp.converter.ByteArrayStxEtxConverter;
import org.springframework.integration.ip.tcp.serializer.ByteArrayStxEtxSerializer;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -194,17 +194,17 @@ public class TcpConfigInboundGatewayTests {
private void stxEtxGuts(Socket socket) throws SocketException, IOException {
socket.setSoTimeout(5000);
String greetings = "Hello World!";
socket.getOutputStream().write(ByteArrayStxEtxConverter.STX);
socket.getOutputStream().write(ByteArrayStxEtxSerializer.STX);
socket.getOutputStream().write((greetings).getBytes());
socket.getOutputStream().write(ByteArrayStxEtxConverter.ETX);
socket.getOutputStream().write(ByteArrayStxEtxSerializer.ETX);
StringBuilder sb = new StringBuilder();
int c;
while (true) {
c = socket.getInputStream().read();
if (c == ByteArrayStxEtxConverter.STX) {
if (c == ByteArrayStxEtxSerializer.STX) {
continue;
}
if (c == ByteArrayStxEtxConverter.ETX) {
if (c == ByteArrayStxEtxSerializer.ETX) {
break;
}
sb.append((char) c);

56
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java
View File

@@ -47,7 +47,7 @@ import org.springframework.integration.ip.tcp.connection.TcpConnectionIntercepto
import org.springframework.integration.ip.tcp.connection.TcpConnectionInterceptorFactoryChain;
import org.springframework.integration.ip.tcp.connection.TcpNetServerConnectionFactory;
import org.springframework.integration.ip.tcp.connection.TcpNioServerConnectionFactory;
import org.springframework.integration.ip.tcp.converter.ByteArrayCrLfConverter;
import org.springframework.integration.ip.tcp.serializer.ByteArrayCrLfSerializer;
import org.springframework.integration.ip.util.SocketUtils;
/**
@@ -59,9 +59,9 @@ public class TcpReceivingChannelAdapterTests {
public void newTestNet() throws Exception {
final int port = SocketUtils.findAvailableServerSocket();
AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(port);
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
scf.setSerializer(converter);
scf.setDeserializer(converter);
ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
scf.setSerializer(serializer);
scf.setDeserializer(serializer);
TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter();
adapter.setConnectionFactory(scf);
scf.start();
@@ -89,9 +89,9 @@ public class TcpReceivingChannelAdapterTests {
public void newTestNio() throws Exception {
final int port = SocketUtils.findAvailableServerSocket();
TcpNioServerConnectionFactory scf = new TcpNioServerConnectionFactory(port);
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
scf.setSerializer(converter);
scf.setDeserializer(converter);
ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
scf.setSerializer(serializer);
scf.setDeserializer(serializer);
scf.setSoTimeout(5000);
TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter();
adapter.setConnectionFactory(scf);
@@ -124,9 +124,9 @@ public class TcpReceivingChannelAdapterTests {
public void newTestNetShared() throws Exception {
final int port = SocketUtils.findAvailableServerSocket();
AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(port);
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
scf.setSerializer(converter);
scf.setDeserializer(converter);
ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
scf.setSerializer(serializer);
scf.setDeserializer(serializer);
TcpSendingMessageHandler handler = new TcpSendingMessageHandler();
handler.setConnectionFactory(scf);
TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter();
@@ -162,9 +162,9 @@ public class TcpReceivingChannelAdapterTests {
public void newTestNioShared() throws Exception {
final int port = SocketUtils.findAvailableServerSocket();
TcpNioServerConnectionFactory scf = new TcpNioServerConnectionFactory(port);
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
scf.setSerializer(converter);
scf.setDeserializer(converter);
ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
scf.setSerializer(serializer);
scf.setDeserializer(serializer);
TcpSendingMessageHandler handler = new TcpSendingMessageHandler();
handler.setConnectionFactory(scf);
TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter();
@@ -200,9 +200,9 @@ public class TcpReceivingChannelAdapterTests {
public void newTestNetSingleNoOutbound() throws Exception {
final int port = SocketUtils.findAvailableServerSocket();
AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(port);
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
scf.setSerializer(converter);
scf.setDeserializer(converter);
ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
scf.setSerializer(serializer);
scf.setDeserializer(serializer);
scf.setSingleUse(true);
TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter();
adapter.setConnectionFactory(scf);
@@ -236,9 +236,9 @@ public class TcpReceivingChannelAdapterTests {
public void newTestNioSingleNoOutbound() throws Exception {
final int port = SocketUtils.findAvailableServerSocket();
TcpNioServerConnectionFactory scf = new TcpNioServerConnectionFactory(port);
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
scf.setSerializer(converter);
scf.setDeserializer(converter);
ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
scf.setSerializer(serializer);
scf.setDeserializer(serializer);
scf.setSingleUse(true);
TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter();
adapter.setConnectionFactory(scf);
@@ -282,9 +282,9 @@ public class TcpReceivingChannelAdapterTests {
public void newTestNetSingleShared() throws Exception {
final int port = SocketUtils.findAvailableServerSocket();
AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(port);
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
scf.setSerializer(converter);
scf.setDeserializer(converter);
ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
scf.setSerializer(serializer);
scf.setDeserializer(serializer);
scf.setSingleUse(true);
TcpSendingMessageHandler handler = new TcpSendingMessageHandler();
handler.setConnectionFactory(scf);
@@ -323,9 +323,9 @@ public class TcpReceivingChannelAdapterTests {
public void newTestNioSingleShared() throws Exception {
final int port = SocketUtils.findAvailableServerSocket();
TcpNioServerConnectionFactory scf = new TcpNioServerConnectionFactory(port);
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
scf.setSerializer(converter);
scf.setDeserializer(converter);
ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
scf.setSerializer(serializer);
scf.setDeserializer(serializer);
scf.setSingleUse(true);
TcpSendingMessageHandler handler = new TcpSendingMessageHandler();
handler.setConnectionFactory(scf);
@@ -364,9 +364,9 @@ public class TcpReceivingChannelAdapterTests {
public void newTestNioSingleSharedMany() throws Exception {
final int port = SocketUtils.findAvailableServerSocket();
TcpNioServerConnectionFactory scf = new TcpNioServerConnectionFactory(port);
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
scf.setSerializer(converter);
scf.setDeserializer(converter);
ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
scf.setSerializer(serializer);
scf.setDeserializer(serializer);
scf.setSingleUse(true);
scf.setPoolSize(100);
TcpSendingMessageHandler handler = new TcpSendingMessageHandler();

72
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java
View File

@@ -50,9 +50,9 @@ import org.springframework.integration.ip.tcp.connection.TcpConnectionIntercepto
import org.springframework.integration.ip.tcp.connection.TcpConnectionInterceptorFactoryChain;
import org.springframework.integration.ip.tcp.connection.TcpNetClientConnectionFactory;
import org.springframework.integration.ip.tcp.connection.TcpNioClientConnectionFactory;
import org.springframework.integration.ip.tcp.converter.ByteArrayCrLfConverter;
import org.springframework.integration.ip.tcp.converter.ByteArrayLengthHeaderConverter;
import org.springframework.integration.ip.tcp.converter.ByteArrayStxEtxConverter;
import org.springframework.integration.ip.tcp.serializer.ByteArrayCrLfSerializer;
import org.springframework.integration.ip.tcp.serializer.ByteArrayLengthHeaderSerializer;
import org.springframework.integration.ip.tcp.serializer.ByteArrayStxEtxSerializer;
import org.springframework.integration.ip.util.SocketUtils;
import org.springframework.integration.support.MessageBuilder;
@@ -94,9 +94,9 @@ public class TcpSendingMessageHandlerTests {
}
});
AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port);
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
ccf.setSerializer(converter);
ccf.setDeserializer(converter);
ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
ccf.setSerializer(serializer);
ccf.setDeserializer(serializer);
ccf.setSoTimeout(10000);
TcpSendingMessageHandler handler = new TcpSendingMessageHandler();
handler.setConnectionFactory(ccf);
@@ -142,9 +142,9 @@ public class TcpSendingMessageHandlerTests {
}
});
AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port);
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
ccf.setSerializer(converter);
ccf.setDeserializer(converter);
ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
ccf.setSerializer(serializer);
ccf.setDeserializer(serializer);
ccf.setSoTimeout(10000);
ccf.start();
TcpSendingMessageHandler handler = new TcpSendingMessageHandler();
@@ -194,9 +194,9 @@ public class TcpSendingMessageHandlerTests {
}
});
AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port);
ByteArrayStxEtxConverter converter = new ByteArrayStxEtxConverter();
ccf.setSerializer(converter);
ccf.setDeserializer(converter);
ByteArrayStxEtxSerializer serializer = new ByteArrayStxEtxSerializer();
ccf.setSerializer(serializer);
ccf.setDeserializer(serializer);
ccf.setSoTimeout(10000);
ccf.start();
TcpSendingMessageHandler handler = new TcpSendingMessageHandler();
@@ -243,9 +243,9 @@ public class TcpSendingMessageHandlerTests {
}
});
AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port);
ByteArrayStxEtxConverter converter = new ByteArrayStxEtxConverter();
ccf.setSerializer(converter);
ccf.setDeserializer(converter);
ByteArrayStxEtxSerializer serializer = new ByteArrayStxEtxSerializer();
ccf.setSerializer(serializer);
ccf.setDeserializer(serializer);
ccf.setSoTimeout(10000);
ccf.start();
TcpSendingMessageHandler handler = new TcpSendingMessageHandler();
@@ -298,9 +298,9 @@ public class TcpSendingMessageHandlerTests {
}
});
AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port);
ByteArrayLengthHeaderConverter converter = new ByteArrayLengthHeaderConverter();
ccf.setSerializer(converter);
ccf.setDeserializer(converter);
ByteArrayLengthHeaderSerializer serializer = new ByteArrayLengthHeaderSerializer();
ccf.setSerializer(serializer);
ccf.setDeserializer(serializer);
ccf.setSoTimeout(10000);
ccf.start();
TcpSendingMessageHandler handler = new TcpSendingMessageHandler();
@@ -350,9 +350,9 @@ public class TcpSendingMessageHandlerTests {
}
});
AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port);
ByteArrayLengthHeaderConverter converter = new ByteArrayLengthHeaderConverter();
ccf.setSerializer(converter);
ccf.setDeserializer(converter);
ByteArrayLengthHeaderSerializer serializer = new ByteArrayLengthHeaderSerializer();
ccf.setSerializer(serializer);
ccf.setDeserializer(serializer);
ccf.setSoTimeout(10000);
ccf.start();
TcpSendingMessageHandler handler = new TcpSendingMessageHandler();
@@ -501,9 +501,9 @@ public class TcpSendingMessageHandlerTests {
}
});
AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port);
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
ccf.setSerializer(converter);
ccf.setDeserializer(converter);
ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
ccf.setSerializer(serializer);
ccf.setDeserializer(serializer);
ccf.setSoTimeout(10000);
ccf.start();
ccf.setSingleUse(true);
@@ -542,9 +542,9 @@ public class TcpSendingMessageHandlerTests {
}
});
AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port);
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
ccf.setSerializer(converter);
ccf.setDeserializer(converter);
ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
ccf.setSerializer(serializer);
ccf.setDeserializer(serializer);
ccf.setSoTimeout(10000);
ccf.start();
ccf.setSingleUse(true);
@@ -585,9 +585,9 @@ public class TcpSendingMessageHandlerTests {
}
});
AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port);
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
ccf.setSerializer(converter);
ccf.setDeserializer(converter);
ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
ccf.setSerializer(serializer);
ccf.setDeserializer(serializer);
ccf.setSoTimeout(10000);
ccf.start();
ccf.setSingleUse(true);
@@ -640,9 +640,9 @@ public class TcpSendingMessageHandlerTests {
}
});
AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port);
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
ccf.setSerializer(converter);
ccf.setDeserializer(converter);
ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
ccf.setSerializer(serializer);
ccf.setDeserializer(serializer);
ccf.setSoTimeout(10000);
ccf.start();
ccf.setSingleUse(true);
@@ -695,9 +695,9 @@ public class TcpSendingMessageHandlerTests {
}
});
AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port);
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
ccf.setSerializer(converter);
ccf.setDeserializer(converter);
ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
ccf.setSerializer(serializer);
ccf.setDeserializer(serializer);
ccf.setSoTimeout(10000);
ccf.setSingleUse(true);
ccf.setTaskExecutor(Executors.newFixedThreadPool(100));

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

@@ -12,29 +12,29 @@
<bean id="tcpIpUtils" class="org.springframework.integration.ip.util.SocketUtils" />
<bean id="crLfConverter" class="org.springframework.integration.ip.tcp.converter.ByteArrayCrLfConverter" />
<bean id="stxEtxConverter" class="org.springframework.integration.ip.tcp.converter.ByteArrayStxEtxConverter" />
<bean id="lengthHeaderConverter" class="org.springframework.integration.ip.tcp.converter.ByteArrayLengthHeaderConverter" />
<bean id="crLfSerializer" class="org.springframework.integration.ip.tcp.serializer.ByteArrayCrLfSerializer" />
<bean id="stxEtxSerializer" class="org.springframework.integration.ip.tcp.serializer.ByteArrayStxEtxSerializer" />
<bean id="lengthHeaderSerializer" class="org.springframework.integration.ip.tcp.serializer.ByteArrayLengthHeaderSerializer" />
<bean id="javaSerializer" class="org.springframework.commons.serializer.DefaultSerializer" />
<bean id="javaDeserializer" class="org.springframework.commons.serializer.DefaultDeserializer" />
<ip:tcp-connection-factory id="crLfServer"
type="server"
port="#{tcpIpUtils.findAvailableServerSocket(7000)}"
serializer="crLfConverter"
deserializer="crLfConverter"/>
serializer="crLfSerializer"
deserializer="crLfSerializer"/>
<ip:tcp-connection-factory id="stxEtxServer"
type="server"
port="#{tcpIpUtils.findAvailableServerSocket(7100)}"
serializer="stxEtxConverter"
deserializer="stxEtxConverter"/>
serializer="stxEtxSerializer"
deserializer="stxEtxSerializer"/>
<ip:tcp-connection-factory id="lengthHeaderServer"
type="server"
port="#{tcpIpUtils.findAvailableServerSocket(7200)}"
serializer="lengthHeaderConverter"
deserializer="lengthHeaderConverter"/>
serializer="lengthHeaderSerializer"
deserializer="lengthHeaderSerializer"/>
<ip:tcp-connection-factory id="javaSerialServer"
type="server"
@@ -46,22 +46,22 @@
type="client"
host="localhost"
port="#{crLfServer.port}"
serializer="crLfConverter"
deserializer="crLfConverter"/>
serializer="crLfSerializer"
deserializer="crLfSerializer"/>
<ip:tcp-connection-factory id="stxEtxClient"
type="client"
host="localhost"
port="#{stxEtxServer.port}"
serializer="stxEtxConverter"
deserializer="stxEtxConverter"/>
serializer="stxEtxSerializer"
deserializer="stxEtxSerializer"/>
<ip:tcp-connection-factory id="lengthHeaderClient"
type="client"
host="localhost"
port="#{lengthHeaderServer.port}"
serializer="lengthHeaderConverter"
deserializer="lengthHeaderConverter"/>
serializer="lengthHeaderSerializer"
deserializer="lengthHeaderSerializer"/>
<ip:tcp-connection-factory id="javaSerialClient"
type="client"
@@ -73,22 +73,22 @@
<ip:tcp-connection-factory id="crLfServerNio"
type="server"
port="#{tcpIpUtils.findAvailableServerSocket(7400)}"
serializer="crLfConverter"
deserializer="crLfConverter"
serializer="crLfSerializer"
deserializer="crLfSerializer"
using-nio="true"/>
<ip:tcp-connection-factory id="stxEtxServerNio"
type="server"
port="#{tcpIpUtils.findAvailableServerSocket(7500)}"
serializer="stxEtxConverter"
deserializer="stxEtxConverter"
serializer="stxEtxSerializer"
deserializer="stxEtxSerializer"
using-nio="true"/>
<ip:tcp-connection-factory id="lengthHeaderServerNio"
type="server"
port="#{tcpIpUtils.findAvailableServerSocket(7600)}"
serializer="lengthHeaderConverter"
deserializer="lengthHeaderConverter"
serializer="lengthHeaderSerializer"
deserializer="lengthHeaderSerializer"
using-nio="true"/>
<ip:tcp-connection-factory id="javaSerialServerNio"
@@ -102,24 +102,24 @@
type="client"
host="localhost"
port="#{crLfServer.port}"
serializer="crLfConverter"
deserializer="crLfConverter"
serializer="crLfSerializer"
deserializer="crLfSerializer"
using-nio="true"/>
<ip:tcp-connection-factory id="stxEtxClientNio"
type="client"
host="localhost"
port="#{stxEtxServer.port}"
serializer="stxEtxConverter"
deserializer="stxEtxConverter"
serializer="stxEtxSerializer"
deserializer="stxEtxSerializer"
using-nio="true"/>
<ip:tcp-connection-factory id="lengthHeaderClientNio"
type="client"
host="localhost"
port="#{lengthHeaderServer.port}"
serializer="lengthHeaderConverter"
deserializer="lengthHeaderConverter"
serializer="lengthHeaderSerializer"
deserializer="lengthHeaderSerializer"
using-nio="true"/>
<ip:tcp-connection-factory id="javaSerialClientNio"

72
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionReadTests.java
View File

@@ -34,10 +34,10 @@ import javax.net.SocketFactory;
import org.junit.Test;
import org.springframework.integration.Message;
import org.springframework.integration.ip.tcp.converter.AbstractByteArrayStreamingConverter;
import org.springframework.integration.ip.tcp.converter.ByteArrayCrLfConverter;
import org.springframework.integration.ip.tcp.converter.ByteArrayLengthHeaderConverter;
import org.springframework.integration.ip.tcp.converter.ByteArrayStxEtxConverter;
import org.springframework.integration.ip.tcp.serializer.AbstractByteArraySerializer;
import org.springframework.integration.ip.tcp.serializer.ByteArrayCrLfSerializer;
import org.springframework.integration.ip.tcp.serializer.ByteArrayLengthHeaderSerializer;
import org.springframework.integration.ip.tcp.serializer.ByteArrayStxEtxSerializer;
import org.springframework.integration.ip.util.SocketUtils;
/**
@@ -49,15 +49,15 @@ public class TcpNioConnectionReadTests {
private CountDownLatch latch = new CountDownLatch(1);
private AbstractServerConnectionFactory getConnectionFactory(int port,
AbstractByteArrayStreamingConverter converter, TcpListener listener) throws Exception {
return getConnectionFactory(port, converter, listener, null);
AbstractByteArraySerializer serializer, TcpListener listener) throws Exception {
return getConnectionFactory(port, serializer, listener, null);
}
private AbstractServerConnectionFactory getConnectionFactory(int port,
AbstractByteArrayStreamingConverter converter, TcpListener listener, TcpSender sender) throws Exception {
AbstractByteArraySerializer serializer, TcpListener listener, TcpSender sender) throws Exception {
AbstractServerConnectionFactory scf = new TcpNioServerConnectionFactory(port);
scf.setSerializer(converter);
scf.setDeserializer(converter);
scf.setSerializer(serializer);
scf.setDeserializer(serializer);
scf.registerListener(listener);
if (sender != null) {
scf.registerSender(sender);
@@ -80,10 +80,10 @@ public class TcpNioConnectionReadTests {
@Test
public void testReadLength() throws Exception {
int port = SocketUtils.findAvailableServerSocket();
ByteArrayLengthHeaderConverter converter = new ByteArrayLengthHeaderConverter();
ByteArrayLengthHeaderSerializer serializer = new ByteArrayLengthHeaderSerializer();
final List<Message<?>> responses = new ArrayList<Message<?>>();
final Semaphore semaphore = new Semaphore(0);
AbstractServerConnectionFactory scf = getConnectionFactory(port, converter,new TcpListener() {
AbstractServerConnectionFactory scf = getConnectionFactory(port, serializer,new TcpListener() {
public boolean onMessage(Message<?> message) {
responses.add(message);
semaphore.release();
@@ -111,10 +111,10 @@ public class TcpNioConnectionReadTests {
@Test
public void testFragmented() throws Exception {
int port = SocketUtils.findAvailableServerSocket();
ByteArrayLengthHeaderConverter converter = new ByteArrayLengthHeaderConverter();
ByteArrayLengthHeaderSerializer serializer = new ByteArrayLengthHeaderSerializer();
final List<Message<?>> responses = new ArrayList<Message<?>>();
final Semaphore semaphore = new Semaphore(0);
AbstractServerConnectionFactory scf = getConnectionFactory(port, converter,new TcpListener() {
AbstractServerConnectionFactory scf = getConnectionFactory(port, serializer,new TcpListener() {
public boolean onMessage(Message<?> message) {
responses.add(message);
try {
@@ -145,10 +145,10 @@ public class TcpNioConnectionReadTests {
@Test
public void testReadStxEtx() throws Exception {
int port = SocketUtils.findAvailableServerSocket();
ByteArrayStxEtxConverter converter = new ByteArrayStxEtxConverter();
ByteArrayStxEtxSerializer serializer = new ByteArrayStxEtxSerializer();
final List<Message<?>> responses = new ArrayList<Message<?>>();
final Semaphore semaphore = new Semaphore(0);
AbstractServerConnectionFactory scf = getConnectionFactory(port, converter,new TcpListener() {
AbstractServerConnectionFactory scf = getConnectionFactory(port, serializer,new TcpListener() {
public boolean onMessage(Message<?> message) {
responses.add(message);
semaphore.release();
@@ -177,10 +177,10 @@ public class TcpNioConnectionReadTests {
@Test
public void testReadCrLf() throws Exception {
int port = SocketUtils.findAvailableServerSocket();
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
final List<Message<?>> responses = new ArrayList<Message<?>>();
final Semaphore semaphore = new Semaphore(0);
AbstractServerConnectionFactory scf = getConnectionFactory(port, converter,new TcpListener() {
AbstractServerConnectionFactory scf = getConnectionFactory(port, serializer,new TcpListener() {
public boolean onMessage(Message<?> message) {
responses.add(message);
semaphore.release();
@@ -208,12 +208,12 @@ public class TcpNioConnectionReadTests {
@Test
public void testReadLengthOverflow() throws Exception {
int port = SocketUtils.findAvailableServerSocket();
ByteArrayLengthHeaderConverter converter = new ByteArrayLengthHeaderConverter();
ByteArrayLengthHeaderSerializer serializer = new ByteArrayLengthHeaderSerializer();
final List<Message<?>> responses = new ArrayList<Message<?>>();
final Semaphore semaphore = new Semaphore(0);
final List<TcpConnection> added = new ArrayList<TcpConnection>();
final List<TcpConnection> removed = new ArrayList<TcpConnection>();
AbstractServerConnectionFactory scf = getConnectionFactory(port, converter,new TcpListener() {
AbstractServerConnectionFactory scf = getConnectionFactory(port, serializer,new TcpListener() {
public boolean onMessage(Message<?> message) {
responses.add(message);
semaphore.release();
@@ -245,13 +245,13 @@ public class TcpNioConnectionReadTests {
@Test
public void testReadStxEtxOverflow() throws Exception {
int port = SocketUtils.findAvailableServerSocket();
ByteArrayStxEtxConverter converter = new ByteArrayStxEtxConverter();
converter.setMaxMessageSize(1024);
ByteArrayStxEtxSerializer serializer = new ByteArrayStxEtxSerializer();
serializer.setMaxMessageSize(1024);
final List<Message<?>> responses = new ArrayList<Message<?>>();
final Semaphore semaphore = new Semaphore(0);
final List<TcpConnection> added = new ArrayList<TcpConnection>();
final List<TcpConnection> removed = new ArrayList<TcpConnection>();
AbstractServerConnectionFactory scf = getConnectionFactory(port, converter,new TcpListener() {
AbstractServerConnectionFactory scf = getConnectionFactory(port, serializer,new TcpListener() {
public boolean onMessage(Message<?> message) {
responses.add(message);
semaphore.release();
@@ -283,13 +283,13 @@ public class TcpNioConnectionReadTests {
@Test
public void testReadCrLfOverflow() throws Exception {
int port = SocketUtils.findAvailableServerSocket();
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
converter.setMaxMessageSize(1024);
ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
serializer.setMaxMessageSize(1024);
final List<Message<?>> responses = new ArrayList<Message<?>>();
final Semaphore semaphore = new Semaphore(0);
final List<TcpConnection> added = new ArrayList<TcpConnection>();
final List<TcpConnection> removed = new ArrayList<TcpConnection>();
AbstractServerConnectionFactory scf = getConnectionFactory(port, converter,new TcpListener() {
AbstractServerConnectionFactory scf = getConnectionFactory(port, serializer,new TcpListener() {
public boolean onMessage(Message<?> message) {
responses.add(message);
semaphore.release();
@@ -323,13 +323,13 @@ public class TcpNioConnectionReadTests {
@Test
public void testCloseCleanupNoData() throws Exception {
int port = SocketUtils.findAvailableServerSocket();
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
converter.setMaxMessageSize(1024);
ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
serializer.setMaxMessageSize(1024);
final List<Message<?>> responses = new ArrayList<Message<?>>();
final Semaphore semaphore = new Semaphore(0);
final List<TcpConnection> added = new ArrayList<TcpConnection>();
final List<TcpConnection> removed = new ArrayList<TcpConnection>();
AbstractServerConnectionFactory scf = getConnectionFactory(port, converter,new TcpListener() {
AbstractServerConnectionFactory scf = getConnectionFactory(port, serializer,new TcpListener() {
public boolean onMessage(Message<?> message) {
responses.add(message);
semaphore.release();
@@ -360,8 +360,8 @@ public class TcpNioConnectionReadTests {
*/
@Test
public void testCloseCleanupCrLf() throws Exception {
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
testClosureMidMessageGuts(converter, "xx");
ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
testClosureMidMessageGuts(serializer, "xx");
}
/**
@@ -372,8 +372,8 @@ public class TcpNioConnectionReadTests {
@Test
public void testCloseCleanupStxEtx() throws Exception {
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
testClosureMidMessageGuts(converter, ByteArrayStxEtxConverter.STX + "xx");
ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
testClosureMidMessageGuts(serializer, ByteArrayStxEtxSerializer.STX + "xx");
}
/**
@@ -384,11 +384,11 @@ public class TcpNioConnectionReadTests {
@Test
public void testCloseCleanupLengthHeader() throws Exception {
ByteArrayLengthHeaderConverter converter = new ByteArrayLengthHeaderConverter();
testClosureMidMessageGuts(converter, "\u0000\u0000\u0000\u0003xx");
ByteArrayLengthHeaderSerializer serializer = new ByteArrayLengthHeaderSerializer();
testClosureMidMessageGuts(serializer, "\u0000\u0000\u0000\u0003xx");
}
private void testClosureMidMessageGuts(AbstractByteArrayStreamingConverter converter, String shortMessage)
private void testClosureMidMessageGuts(AbstractByteArraySerializer serializer, String shortMessage)
throws Exception, IOException, UnknownHostException,
InterruptedException {
final int port = SocketUtils.findAvailableServerSocket();
@@ -396,7 +396,7 @@ public class TcpNioConnectionReadTests {
final Semaphore semaphore = new Semaphore(0);
final List<TcpConnection> added = new ArrayList<TcpConnection>();
final List<TcpConnection> removed = new ArrayList<TcpConnection>();
AbstractServerConnectionFactory scf = getConnectionFactory(port, converter,new TcpListener() {
AbstractServerConnectionFactory scf = getConnectionFactory(port, serializer,new TcpListener() {
public boolean onMessage(Message<?> message) {
responses.add(message);
return false;

46
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionWriteTests.java
View File

@@ -28,10 +28,10 @@ import javax.net.ServerSocketFactory;
import org.junit.Test;
import org.springframework.integration.ip.tcp.converter.AbstractByteArrayStreamingConverter;
import org.springframework.integration.ip.tcp.converter.ByteArrayCrLfConverter;
import org.springframework.integration.ip.tcp.converter.ByteArrayLengthHeaderConverter;
import org.springframework.integration.ip.tcp.converter.ByteArrayStxEtxConverter;
import org.springframework.integration.ip.tcp.serializer.AbstractByteArraySerializer;
import org.springframework.integration.ip.tcp.serializer.ByteArrayCrLfSerializer;
import org.springframework.integration.ip.tcp.serializer.ByteArrayLengthHeaderSerializer;
import org.springframework.integration.ip.tcp.serializer.ByteArrayStxEtxSerializer;
import org.springframework.integration.ip.util.SocketUtils;
import org.springframework.integration.support.MessageBuilder;
@@ -42,10 +42,10 @@ import org.springframework.integration.support.MessageBuilder;
public class TcpNioConnectionWriteTests {
private AbstractConnectionFactory getClientConnectionFactory(boolean direct,
final int port, AbstractByteArrayStreamingConverter converter) {
final int port, AbstractByteArraySerializer serializer) {
TcpNioClientConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port);
ccf.setSerializer(converter);
ccf.setDeserializer(converter);
ccf.setSerializer(serializer);
ccf.setDeserializer(serializer);
ccf.setSoTimeout(10000);
ccf.setUsingDirectBuffers(direct);
ccf.start();
@@ -62,8 +62,8 @@ public class TcpNioConnectionWriteTests {
Thread t = new Thread(new Runnable() {
public void run() {
try {
ByteArrayLengthHeaderConverter converter = new ByteArrayLengthHeaderConverter();
AbstractConnectionFactory ccf = getClientConnectionFactory(false, port, converter);
ByteArrayLengthHeaderSerializer serializer = new ByteArrayLengthHeaderSerializer();
AbstractConnectionFactory ccf = getClientConnectionFactory(false, port, serializer);
TcpConnection connection = ccf.getConnection();
connection.send(MessageBuilder.withPayload(testString.getBytes()).build());
Thread.sleep(1000000000L);
@@ -95,8 +95,8 @@ public class TcpNioConnectionWriteTests {
Thread t = new Thread(new Runnable() {
public void run() {
try {
ByteArrayStxEtxConverter converter = new ByteArrayStxEtxConverter();
AbstractConnectionFactory ccf = getClientConnectionFactory(false, port, converter);
ByteArrayStxEtxSerializer serializer = new ByteArrayStxEtxSerializer();
AbstractConnectionFactory ccf = getClientConnectionFactory(false, port, serializer);
TcpConnection connection = ccf.getConnection();
connection.send(MessageBuilder.withPayload(testString.getBytes()).build());
Thread.sleep(1000000000L);
@@ -112,9 +112,9 @@ public class TcpNioConnectionWriteTests {
InputStream is = socket.getInputStream();
byte[] buff = new byte[testString.length() + 2];
readFully(is, buff);
assertEquals(ByteArrayStxEtxConverter.STX, buff[0]);
assertEquals(ByteArrayStxEtxSerializer.STX, buff[0]);
assertEquals(testString, new String(buff, 1, testString.length()));
assertEquals(ByteArrayStxEtxConverter.ETX, buff[testString.length() + 1]);
assertEquals(ByteArrayStxEtxSerializer.ETX, buff[testString.length() + 1]);
server.close();
}
@@ -128,8 +128,8 @@ public class TcpNioConnectionWriteTests {
Thread t = new Thread(new Runnable() {
public void run() {
try {
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
AbstractConnectionFactory ccf = getClientConnectionFactory(false, port, converter);
ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
AbstractConnectionFactory ccf = getClientConnectionFactory(false, port, serializer);
TcpConnection connection = ccf.getConnection();
connection.send(MessageBuilder.withPayload(testString.getBytes()).build());
Thread.sleep(1000000000L);
@@ -161,8 +161,8 @@ public class TcpNioConnectionWriteTests {
Thread t = new Thread(new Runnable() {
public void run() {
try {
ByteArrayLengthHeaderConverter converter = new ByteArrayLengthHeaderConverter();
AbstractConnectionFactory ccf = getClientConnectionFactory(true, port, converter);
ByteArrayLengthHeaderSerializer serializer = new ByteArrayLengthHeaderSerializer();
AbstractConnectionFactory ccf = getClientConnectionFactory(true, port, serializer);
TcpConnection connection = ccf.getConnection();
connection.send(MessageBuilder.withPayload(testString.getBytes()).build());
Thread.sleep(1000000000L);
@@ -194,8 +194,8 @@ public class TcpNioConnectionWriteTests {
Thread t = new Thread(new Runnable() {
public void run() {
try {
ByteArrayStxEtxConverter converter = new ByteArrayStxEtxConverter();
AbstractConnectionFactory ccf = getClientConnectionFactory(true, port, converter);
ByteArrayStxEtxSerializer serializer = new ByteArrayStxEtxSerializer();
AbstractConnectionFactory ccf = getClientConnectionFactory(true, port, serializer);
TcpConnection connection = ccf.getConnection();
connection.send(MessageBuilder.withPayload(testString.getBytes()).build());
Thread.sleep(1000000000L);
@@ -212,9 +212,9 @@ public class TcpNioConnectionWriteTests {
InputStream is = socket.getInputStream();
byte[] buff = new byte[testString.length() + 2];
readFully(is, buff);
assertEquals(ByteArrayStxEtxConverter.STX, buff[0]);
assertEquals(ByteArrayStxEtxSerializer.STX, buff[0]);
assertEquals(testString, new String(buff, 1, testString.length()));
assertEquals(ByteArrayStxEtxConverter.ETX, buff[testString.length() + 1]);
assertEquals(ByteArrayStxEtxSerializer.ETX, buff[testString.length() + 1]);
server.close();
}
@@ -228,8 +228,8 @@ public class TcpNioConnectionWriteTests {
Thread t = new Thread(new Runnable() {
public void run() {
try {
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
AbstractConnectionFactory ccf = getClientConnectionFactory(true, port, converter);
ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
AbstractConnectionFactory ccf = getClientConnectionFactory(true, port, serializer);
TcpConnection connection = ccf.getConnection();
connection.send(MessageBuilder.withPayload(testString.getBytes()).build());
Thread.sleep(1000000000L);

47
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/converter/DeserializationTests.java → spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/serializer/DeserializationTests.java
View File

@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.integration.ip.tcp.converter;
package org.springframework.integration.ip.tcp.serializer;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.fail;
@@ -28,6 +28,9 @@ import javax.net.ServerSocketFactory;
import org.junit.Test;
import org.springframework.commons.serializer.DefaultDeserializer;
import org.springframework.integration.ip.tcp.serializer.ByteArrayCrLfSerializer;
import org.springframework.integration.ip.tcp.serializer.ByteArrayLengthHeaderSerializer;
import org.springframework.integration.ip.tcp.serializer.ByteArrayStxEtxSerializer;
import org.springframework.integration.ip.util.SocketUtils;
/**
@@ -44,11 +47,11 @@ public class DeserializationTests {
SocketUtils.testSendLength(port, null);
Socket socket = server.accept();
socket.setSoTimeout(5000);
ByteArrayLengthHeaderConverter converter = new ByteArrayLengthHeaderConverter();
byte[] out = converter.deserialize(socket.getInputStream());
ByteArrayLengthHeaderSerializer serializer = new ByteArrayLengthHeaderSerializer();
byte[] out = serializer.deserialize(socket.getInputStream());
assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING,
new String(out));
out = converter.deserialize(socket.getInputStream());
out = serializer.deserialize(socket.getInputStream());
assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING,
new String(out));
server.close();
@@ -62,11 +65,11 @@ public class DeserializationTests {
SocketUtils.testSendStxEtx(port, null);
Socket socket = server.accept();
socket.setSoTimeout(5000);
ByteArrayStxEtxConverter converter = new ByteArrayStxEtxConverter();
byte[] out = converter.deserialize(socket.getInputStream());
ByteArrayStxEtxSerializer serializer = new ByteArrayStxEtxSerializer();
byte[] out = serializer.deserialize(socket.getInputStream());
assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING,
new String(out));
out = converter.deserialize(socket.getInputStream());
out = serializer.deserialize(socket.getInputStream());
assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING,
new String(out));
server.close();
@@ -80,11 +83,11 @@ public class DeserializationTests {
SocketUtils.testSendCrLf(port, null);
Socket socket = server.accept();
socket.setSoTimeout(5000);
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
byte[] out = converter.deserialize(socket.getInputStream());
ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
byte[] out = serializer.deserialize(socket.getInputStream());
assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING,
new String(out));
out = converter.deserialize(socket.getInputStream());
out = serializer.deserialize(socket.getInputStream());
assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING,
new String(out));
server.close();
@@ -114,9 +117,9 @@ public class DeserializationTests {
SocketUtils.testSendLengthOverflow(port);
Socket socket = server.accept();
socket.setSoTimeout(5000);
ByteArrayLengthHeaderConverter converter = new ByteArrayLengthHeaderConverter();
ByteArrayLengthHeaderSerializer serializer = new ByteArrayLengthHeaderSerializer();
try {
converter.deserialize(socket.getInputStream());
serializer.deserialize(socket.getInputStream());
fail("Expected message length exceeded exception");
} catch (IOException e) {
if (!e.getMessage().startsWith("Message length")) {
@@ -135,9 +138,9 @@ public class DeserializationTests {
SocketUtils.testSendStxEtxOverflow(port);
Socket socket = server.accept();
socket.setSoTimeout(500);
ByteArrayStxEtxConverter converter = new ByteArrayStxEtxConverter();
ByteArrayStxEtxSerializer serializer = new ByteArrayStxEtxSerializer();
try {
converter.deserialize(socket.getInputStream());
serializer.deserialize(socket.getInputStream());
fail("Expected timeout exception");
} catch (IOException e) {
if (!e.getMessage().startsWith("Read timed out")) {
@@ -156,10 +159,10 @@ public class DeserializationTests {
SocketUtils.testSendStxEtxOverflow(port);
Socket socket = server.accept();
socket.setSoTimeout(5000);
ByteArrayStxEtxConverter converter = new ByteArrayStxEtxConverter();
converter.setMaxMessageSize(1024);
ByteArrayStxEtxSerializer serializer = new ByteArrayStxEtxSerializer();
serializer.setMaxMessageSize(1024);
try {
converter.deserialize(socket.getInputStream());
serializer.deserialize(socket.getInputStream());
fail("Expected message length exceeded exception");
} catch (IOException e) {
if (!e.getMessage().startsWith("ETX not found")) {
@@ -178,9 +181,9 @@ public class DeserializationTests {
SocketUtils.testSendCrLfOverflow(port);
Socket socket = server.accept();
socket.setSoTimeout(500);
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
try {
converter.deserialize(socket.getInputStream());
serializer.deserialize(socket.getInputStream());
fail("Expected timout exception");
} catch (IOException e) {
if (!e.getMessage().startsWith("Read timed out")) {
@@ -199,10 +202,10 @@ public class DeserializationTests {
SocketUtils.testSendCrLfOverflow(port);
Socket socket = server.accept();
socket.setSoTimeout(5000);
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
converter.setMaxMessageSize(1024);
ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
serializer.setMaxMessageSize(1024);
try {
converter.deserialize(socket.getInputStream());
serializer.deserialize(socket.getInputStream());
fail("Expected message length exceeded exception");
} catch (IOException e) {
if (!e.getMessage().startsWith("CRLF not found")) {

21
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/converter/SerializationTests.java → spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/serializer/SerializationTests.java
View File

@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.integration.ip.tcp.converter;
package org.springframework.integration.ip.tcp.serializer;
import static org.junit.Assert.assertEquals;
@@ -31,6 +31,9 @@ import javax.net.SocketFactory;
import org.junit.Test;
import org.springframework.commons.serializer.DefaultSerializer;
import org.springframework.integration.ip.tcp.serializer.ByteArrayCrLfSerializer;
import org.springframework.integration.ip.tcp.serializer.ByteArrayLengthHeaderSerializer;
import org.springframework.integration.ip.tcp.serializer.ByteArrayStxEtxSerializer;
import org.springframework.integration.ip.util.SocketUtils;
/**
@@ -51,8 +54,8 @@ public class SerializationTests {
Socket socket = SocketFactory.getDefault().createSocket("localhost", port);
ByteBuffer buffer = ByteBuffer.allocate(testString.length());
buffer.put(testString.getBytes());
ByteArrayLengthHeaderConverter converter = new ByteArrayLengthHeaderConverter();
converter.serialize(buffer.array(), socket.getOutputStream());
ByteArrayLengthHeaderSerializer serializer = new ByteArrayLengthHeaderSerializer();
serializer.serialize(buffer.array(), socket.getOutputStream());
Thread.sleep(1000000000L);
} catch (Exception e) {
e.printStackTrace();
@@ -84,8 +87,8 @@ public class SerializationTests {
Socket socket = SocketFactory.getDefault().createSocket("localhost", port);
ByteBuffer buffer = ByteBuffer.allocate(testString.length());
buffer.put(testString.getBytes());
ByteArrayStxEtxConverter converter = new ByteArrayStxEtxConverter();
converter.serialize(buffer.array(), socket.getOutputStream());
ByteArrayStxEtxSerializer serializer = new ByteArrayStxEtxSerializer();
serializer.serialize(buffer.array(), socket.getOutputStream());
Thread.sleep(1000000000L);
} catch (Exception e) {
e.printStackTrace();
@@ -99,9 +102,9 @@ public class SerializationTests {
InputStream is = socket.getInputStream();
byte[] buff = new byte[testString.length() + 2];
readFully(is, buff);
assertEquals(ByteArrayStxEtxConverter.STX, buff[0]);
assertEquals(ByteArrayStxEtxSerializer.STX, buff[0]);
assertEquals(testString, new String(buff, 1, testString.length()));
assertEquals(ByteArrayStxEtxConverter.ETX, buff[testString.length() + 1]);
assertEquals(ByteArrayStxEtxSerializer.ETX, buff[testString.length() + 1]);
server.close();
}
@@ -117,8 +120,8 @@ public class SerializationTests {
Socket socket = SocketFactory.getDefault().createSocket("localhost", port);
ByteBuffer buffer = ByteBuffer.allocate(testString.length());
buffer.put(testString.getBytes());
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
converter.serialize(buffer.array(), socket.getOutputStream());
ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
serializer.serialize(buffer.array(), socket.getOutputStream());
Thread.sleep(1000000000L);
} catch (Exception e) {
e.printStackTrace();