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

INT-1486 updated IP module for refactored Serialization code

Browse Source
This commit is contained in:
Mark Fisher
2010-09-28 15:56:01 -04:00
parent aba3b47060
commit 68648dcb94
28 changed files with 292 additions and 287 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -89,9 +89,9 @@ public abstract class IpAdapterParserUtils {
static final String TCP_CONNECTION_TYPE = "type";
static final String INPUT_CONVERTER = "input-converter";
static final String SERIALIZER = "serializer";
static final String OUTPUT_CONVERTER = "output-converter";
static final String DESERIALIZER = "deserializer";
static final String SINGLE_USE = "single-use";

4
spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionParser.java
View File

@@ -85,9 +85,9 @@ public class TcpConnectionParser extends AbstractBeanDefinitionParser {
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,
IpAdapterParserUtils.TASK_EXECUTOR);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,
IpAdapterParserUtils.INPUT_CONVERTER);
IpAdapterParserUtils.SERIALIZER);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,
IpAdapterParserUtils.OUTPUT_CONVERTER);
IpAdapterParserUtils.DESERIALIZER);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IpAdapterParserUtils.SINGLE_USE);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,

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

@@ -70,8 +70,9 @@ public abstract class AbstractClientConnectionFactory extends AbstractConnection
}
}
connection.setMapper(this.mapper);
connection.setInputConverter(this.inputConverter);
connection.setOutputConverter(this.outputConverter);
connection.setDeserializer(this.deserializer);
connection.setSerializer(this.serializer);
connection.setSingleUse(this.singleUse);
}
}

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

@@ -23,8 +23,9 @@ import java.util.concurrent.Executors;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.commons.serializer.InputStreamingConverter;
import org.springframework.commons.serializer.OutputStreamingConverter;
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.util.Assert;
@@ -67,9 +68,9 @@ public abstract class AbstractConnectionFactory
protected Executor taskExecutor;
protected InputStreamingConverter<?> inputConverter = new ByteArrayCrLfConverter();
protected Deserializer<?> deserializer = new ByteArrayCrLfConverter();
protected OutputStreamingConverter<?> outputConverter = new ByteArrayCrLfConverter();
protected Serializer<?> serializer = new ByteArrayCrLfConverter();
protected TcpMessageMapper mapper = new TcpMessageMapper();
@@ -250,18 +251,18 @@ public abstract class AbstractConnectionFactory
/**
*
* @param converter the inputConverter to set
* @param deserializer the deserializer to set
*/
public void setInputConverter(InputStreamingConverter<?> converter) {
this.inputConverter = converter;
public void setDeserializer(Deserializer<?> deserializer) {
this.deserializer = deserializer;
}
/**
*
* @param outputConverter the outputConverter to set
* @param serializer the serializer to set
*/
public void setOutputConverter(OutputStreamingConverter<?> outputConverter) {
this.outputConverter = outputConverter;
public void setSerializer(Serializer<?> serializer) {
this.serializer = serializer;
}
/**

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

@@ -19,21 +19,21 @@ package org.springframework.integration.ip.tcp.connection;
import java.net.Socket;
import java.net.SocketException;
/**
* Base class for all server connection factories. Server connection factories
* listen on a port for incoming connections and create new TcpConnection objects
* for each new connection.
* for each new connection.
*
* @author Gary Russell
*
* @since 2.0
*/
public abstract class AbstractServerConnectionFactory extends AbstractConnectionFactory {
protected boolean listening;
protected String localAddress;
/**
* The port on which the factory will listen.
* @param port
@@ -41,7 +41,8 @@ public abstract class AbstractServerConnectionFactory extends AbstractConnection
public AbstractServerConnectionFactory(int port) {
this.port = port;
}
/**
* Not supported because the factory manages multiple connections and this
* method cannot discriminate.
@@ -70,8 +71,8 @@ public abstract class AbstractServerConnectionFactory extends AbstractConnection
}
connection.registerSender(this.sender);
connection.setMapper(this.mapper);
connection.setInputConverter(this.inputConverter);
connection.setOutputConverter(this.outputConverter);
connection.setDeserializer(this.deserializer);
connection.setSerializer(this.serializer);
connection.setSingleUse(this.singleUse);
/*
* If we have a collaborating outbound channel adapter and we are configured

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

@@ -20,8 +20,9 @@ import java.util.concurrent.atomic.AtomicLong;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.commons.serializer.InputStreamingConverter;
import org.springframework.commons.serializer.OutputStreamingConverter;
import org.springframework.commons.serializer.Deserializer;
import org.springframework.commons.serializer.Serializer;
import org.springframework.integration.ip.tcp.converter.AbstractByteArrayStreamingConverter;
import org.springframework.util.Assert;
@@ -39,10 +40,10 @@ public abstract class AbstractTcpConnection implements TcpConnection {
protected Log logger = LogFactory.getLog(this.getClass());
@SuppressWarnings("rawtypes")
protected InputStreamingConverter inputConverter;
protected Deserializer deserializer;
@SuppressWarnings("rawtypes")
protected OutputStreamingConverter outputConverter;
protected Serializer serializer;
protected TcpMessageMapper mapper;
@@ -84,42 +85,42 @@ public abstract class AbstractTcpConnection implements TcpConnection {
public void setMapper(TcpMessageMapper mapper) {
Assert.notNull(mapper, this.getClass().getName() + " Mapper may not be null");
this.mapper = mapper;
if (this.outputConverter != null &&
!(this.outputConverter instanceof AbstractByteArrayStreamingConverter)) {
if (this.serializer != null &&
!(this.serializer instanceof AbstractByteArrayStreamingConverter)) {
mapper.setStringToBytes(false);
}
}
/**
*
* @return the input converter
* @return the deserializer
*/
public InputStreamingConverter<?> getInputConverter() {
return inputConverter;
public Deserializer<?> getDeserializer() {
return this.deserializer;
}
/**
* @param inputConverter the input converter to set
* @param deserializer the deserializer to set
*/
public void setInputConverter(InputStreamingConverter<?> inputConverter) {
this.inputConverter = inputConverter;
public void setDeserializer(Deserializer<?> deserializer) {
this.deserializer = deserializer;
}
/**
*
* @return the output converter
* @return the serializer
*/
public OutputStreamingConverter<?> getOutputConverter() {
return outputConverter;
public Serializer<?> getSerializer() {
return this.serializer;
}
/**
* @param outputConverter the output converter to set
* @param serializer the serializer to set
*/
public void setOutputConverter(OutputStreamingConverter<?> outputConverter) {
this.outputConverter = outputConverter;
if (!(outputConverter instanceof AbstractByteArrayStreamingConverter)) {
mapper.setStringToBytes(false);
public void setSerializer(Serializer<?> serializer) {
this.serializer = serializer;
if (!(serializer instanceof AbstractByteArrayStreamingConverter)) {
this.mapper.setStringToBytes(false);
}
}

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

@@ -13,11 +13,11 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.ip.tcp.connection;
import org.springframework.commons.serializer.InputStreamingConverter;
import org.springframework.commons.serializer.OutputStreamingConverter;
import org.springframework.commons.serializer.Deserializer;
import org.springframework.commons.serializer.Serializer;
import org.springframework.integration.Message;
/**
@@ -26,7 +26,6 @@ import org.springframework.integration.Message;
*
* @author Gary Russell
* @since 2.0
*
*/
public abstract class AbstractTcpConnectionInterceptor implements TcpConnectionInterceptor {
@@ -36,6 +35,7 @@ public abstract class AbstractTcpConnectionInterceptor implements TcpConnectionI
private TcpSender tcpSender;
public void close() {
this.theConnection.close();
}
@@ -90,20 +90,20 @@ public abstract class AbstractTcpConnectionInterceptor implements TcpConnectionI
this.theConnection.setMapper(mapper);
}
public InputStreamingConverter<?> getInputConverter() {
return this.theConnection.getInputConverter();
public Deserializer<?> getDeserializer() {
return this.theConnection.getDeserializer();
}
public void setInputConverter(InputStreamingConverter<?> inputConverter) {
this.theConnection.setInputConverter(inputConverter);
public void setDeserializer(Deserializer<?> deserializer) {
this.theConnection.setDeserializer(deserializer);
}
public OutputStreamingConverter<?> getOutputConverter() {
return this.theConnection.getOutputConverter();
public Serializer<?> getSerializer() {
return this.theConnection.getSerializer();
}
public void setOutputConverter(OutputStreamingConverter<?> outputConverter) {
this.theConnection.setOutputConverter(outputConverter);
public void setSerializer(Serializer<?> serializer) {
this.theConnection.setSerializer(serializer);
}
public boolean isServer() {

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

@@ -19,8 +19,8 @@ package org.springframework.integration.ip.tcp.connection;
import java.net.Socket;
import java.nio.channels.SocketChannel;
import org.springframework.commons.serializer.InputStreamingConverter;
import org.springframework.commons.serializer.OutputStreamingConverter;
import org.springframework.commons.serializer.Deserializer;
import org.springframework.commons.serializer.Serializer;
import org.springframework.integration.Message;
/**
@@ -119,25 +119,25 @@ public interface TcpConnection extends Runnable {
/**
*
* @return the input converter
* @return the deserializer
*/
public InputStreamingConverter<?> getInputConverter();
public Deserializer<?> getDeserializer();
/**
* @param inputConverter the inputConverter to set
* @param deserializer the deserializer to set
*/
public void setInputConverter(InputStreamingConverter<?> inputConverter);
public void setDeserializer(Deserializer<?> deserializer);
/**
*
* @return the output converter
* @return the serializer
*/
public OutputStreamingConverter<?> getOutputConverter();
public Serializer<?> getSerializer();
/**
* @param outputConverter the outputConverter to set
* @param serializer the serializer to set
*/
public void setOutputConverter(OutputStreamingConverter<?> outputConverter);
public void setSerializer(Serializer<?> serializer);
/**
* @return this connection's listener

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

@@ -19,7 +19,6 @@ package org.springframework.integration.ip.tcp.connection;
import java.net.Socket;
import java.net.SocketTimeoutException;
import org.springframework.commons.serializer.InputStreamingConverter;
import org.springframework.integration.Message;
import org.springframework.integration.ip.tcp.SocketIoUtils;
import org.springframework.integration.ip.tcp.converter.SoftEndOfStreamException;
@@ -64,7 +63,7 @@ public class TcpNetConnection extends AbstractTcpConnection {
@SuppressWarnings("unchecked")
public synchronized void send(Message<?> message) throws Exception {
Object object = mapper.fromMessage(message);
this.outputConverter.convert(object, this.socket.getOutputStream());
this.serializer.serialize(object, this.socket.getOutputStream());
if (logger.isDebugEnabled())
logger.debug("Message sent " + message);
}
@@ -78,7 +77,7 @@ public class TcpNetConnection extends AbstractTcpConnection {
}
public Object getPayload() throws Exception {
return this.inputConverter.convert(this.socket.getInputStream());
return this.deserializer.deserialize(this.socket.getInputStream());
}
public int getPort() {

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

@@ -106,7 +106,7 @@ public class TcpNioConnection extends AbstractTcpConnection {
public void send(Message<?> message) throws Exception {
synchronized(mapper) {
Object object = mapper.fromMessage(message);
this.outputConverter.convert(object, this.channelOutputStream);
this.serializer.serialize(object, this.channelOutputStream);
}
}
@@ -119,7 +119,7 @@ public class TcpNioConnection extends AbstractTcpConnection {
}
public Object getPayload() throws Exception {
return this.inputConverter.convert(pipedInputStream);
return this.deserializer.deserialize(pipedInputStream);
}
public int getPort() {

9
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/AbstractByteArrayStreamingConverter.java
View File

@@ -20,8 +20,9 @@ import java.io.IOException;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.commons.serializer.InputStreamingConverter;
import org.springframework.commons.serializer.OutputStreamingConverter;
import org.springframework.commons.serializer.Deserializer;
import org.springframework.commons.serializer.Serializer;
/**
* Base class for streaming converters that convert to/from a byte array.
@@ -31,8 +32,8 @@ import org.springframework.commons.serializer.OutputStreamingConverter;
*
*/
public abstract class AbstractByteArrayStreamingConverter implements
InputStreamingConverter<byte[]>,
OutputStreamingConverter<byte[]> {
Serializer<byte[]>,
Deserializer<byte[]> {
protected int maxMessageSize = 2048;

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

@@ -27,7 +27,6 @@ import java.io.OutputStream;
*
* @author Gary Russell
* @since 2.0
*
*/
public class ByteArrayCrLfConverter extends AbstractByteArrayStreamingConverter {
@@ -37,7 +36,7 @@ public class ByteArrayCrLfConverter extends AbstractByteArrayStreamingConverter
* is closed immediately after the \r\n (i.e. no data is in the process of
* being read).
*/
public byte[] convert(InputStream inputStream) throws IOException {
public byte[] deserialize(InputStream inputStream) throws IOException {
byte[] buffer = new byte[this.maxMessageSize];
int n = 0;
int bite;
@@ -66,7 +65,7 @@ public class ByteArrayCrLfConverter extends AbstractByteArrayStreamingConverter
/**
* Writes the byte[] to the stream and appends \r\n.
*/
public void convert(byte[] bytes, OutputStream outputStream) throws IOException {
public void serialize(byte[] bytes, OutputStream outputStream) throws IOException {
outputStream.write(bytes);
outputStream.write('\r');
outputStream.write('\n');

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

@@ -32,8 +32,9 @@ import org.apache.commons.logging.LogFactory;
* The length field contains the length of data following the length
* field.
* (network byte order).
*
* @author Gary Russell
*
* @since 2.0
*/
public class ByteArrayLengthHeaderConverter extends AbstractByteArrayStreamingConverter {
@@ -46,7 +47,7 @@ public class ByteArrayLengthHeaderConverter extends AbstractByteArrayStreamingCo
* Throws a {@link SoftEndOfStreamException} if the stream
* is closed between messages.
*/
public byte[] convert(InputStream inputStream) throws IOException {
public byte[] deserialize(InputStream inputStream) throws IOException {
byte[] lengthPart = new byte[4];
int status = read(inputStream, lengthPart, true);
if (status < 0) {
@@ -69,8 +70,7 @@ public class ByteArrayLengthHeaderConverter extends AbstractByteArrayStreamingCo
* Writes the byte[] to the output stream, preceded by a 4 byte
* length in network byte order (big endian).
*/
public void convert(byte[] bytes, OutputStream outputStream)
throws IOException {
public void serialize(byte[] bytes, OutputStream outputStream) throws IOException {
ByteBuffer lengthPart = ByteBuffer.allocate(4);
lengthPart.putInt(bytes.length);
outputStream.write(lengthPart.array());

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

@@ -29,7 +29,6 @@ import org.springframework.integration.mapping.MessageMappingException;
*
* @author Gary Russell
* @since 2.0
*
*/
public class ByteArrayStxEtxConverter extends AbstractByteArrayStreamingConverter {
@@ -45,7 +44,7 @@ public class ByteArrayStxEtxConverter extends AbstractByteArrayStreamingConverte
* being read).
*
*/
public byte[] convert(InputStream inputStream) throws IOException {
public byte[] deserialize(InputStream inputStream) throws IOException {
int bite = inputStream.read();
if (bite < 0) {
throw new SoftEndOfStreamException("Stream closed between payloads");
@@ -71,7 +70,7 @@ public class ByteArrayStxEtxConverter extends AbstractByteArrayStreamingConverte
* Writes the byte[] to the stream, prefixed by an ASCII STX character and
* terminated with an ASCII ETX character.
*/
public void convert(byte[] bytes, OutputStream outputStream) throws IOException {
public void serialize(byte[] bytes, OutputStream outputStream) throws IOException {
outputStream.write(STX);
outputStream.write(bytes);
outputStream.write(ETX);

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

@@ -290,31 +290,31 @@ the factory, the connection will be closed after a response is received.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="input-converter" type="xsd:string" >
<xsd:attribute name="serializer" type="xsd:string" >
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.commons.serializer.InputStreamingConverter"/>
<tool:expected-type type="org.springframework.commons.serializer.Serializer"/>
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
An InputStreamingConverter that converts message payloads to/from output streams/input streams
associated with the connection. Default is ByteArrayCrLfConverter. Input and output converters
would normally be the same but this is not required.
A Serializer that converts message payloads to/from output streams/input streams
associated with the connection. Default is ByteArrayCrLfConverter. Serializer and Deserializer
would normally be the same but this is not required.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="output-converter" type="xsd:string" >
<xsd:attribute name="deserializer" type="xsd:string" >
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.commons.serializer.OutputStreamingConverter"/>
<tool:expected-type type="org.springframework.commons.serializer.Deserializer"/>
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
An OutputStreamingConverter that converts message payloads to/from output streams/input streams
associated with the connection. Default is ByteArrayCrLfConverter. Input and output converters
would normally be the same but this is not required.
A Deserializer that converts message payloads to/from output streams/input streams
associated with the connection. Default is ByteArrayCrLfConverter. Serializer and Deserializer
would normally be the same but this is not required.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>

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

@@ -131,8 +131,8 @@
type="client"
host="localhost"
port="#{tcpIpUtils.findAvailableServerSocket(6000)}"
input-converter="serial"
output-converter="serial"
serializer="defaultSerializer"
deserializer="defaultDeserializer"
so-keep-alive="true"
so-linger="54"
so-receive-buffer-size="1234"
@@ -153,8 +153,8 @@
type="server"
port="#{client1.port}"
local-address="127.0.0.1"
input-converter="serial"
output-converter="serial"
serializer="defaultSerializer"
deserializer="defaultDeserializer"
so-keep-alive="true"
so-linger="55"
so-receive-buffer-size="1234"
@@ -175,8 +175,8 @@
type="client"
host="localhost"
port="#{tcpIpUtils.findAvailableServerSocket(6100)}"
input-converter="serial"
output-converter="serial"
serializer="defaultSerializer"
deserializer="defaultDeserializer"
so-keep-alive="true"
so-linger="54"
so-receive-buffer-size="1234"
@@ -196,8 +196,8 @@
type="server"
port="#{client1.port}"
local-address="127.0.0.1"
input-converter="serial"
output-converter="serial"
serializer="defaultSerializer"
deserializer="defaultDeserializer"
so-keep-alive="true"
so-linger="55"
so-receive-buffer-size="1234"
@@ -214,8 +214,10 @@
<bean id="interceptors" class="org.springframework.integration.ip.tcp.connection.TcpConnectionInterceptorFactoryChain" />
<bean id="serial" class="org.springframework.commons.serializer.java.JavaStreamingConverter" />
<bean id="defaultSerializer" class="org.springframework.commons.serializer.DefaultSerializer" />
<bean id="defaultDeserializer" class="org.springframework.commons.serializer.DefaultDeserializer" />
<ip:tcp-outbound-channel-adapter id="tcpNewOut1"
channel="tcpChannel"
connection-factory="client1" />

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

@@ -13,6 +13,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.ip.config;
import static org.junit.Assert.assertEquals;
@@ -23,10 +24,12 @@ import static org.junit.Assert.assertTrue;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.commons.serializer.InputStreamingConverter;
import org.springframework.commons.serializer.Deserializer;
import org.springframework.commons.serializer.Serializer;
import org.springframework.context.ApplicationContext;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.ip.tcp.TcpInboundGateway;
@@ -46,7 +49,6 @@ import org.springframework.integration.ip.udp.UnicastSendingMessageHandler;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Gary Russell
* @since 2.0
@@ -111,8 +113,11 @@ public class ParserUnitTests {
AbstractConnectionFactory cfC2;
@Autowired
InputStreamingConverter<byte[]> converter;
Serializer<?> serializer;
@Autowired
Deserializer<?> deserializer;
@Autowired
@Qualifier(value="server1")
AbstractConnectionFactory server1;
@@ -263,8 +268,8 @@ public class ParserUnitTests {
assertEquals(1236, client1.getSoTimeout());
assertEquals(12, client1.getSoTrafficClass());
DirectFieldAccessor dfa = new DirectFieldAccessor(client1);
assertSame(converter, dfa.getPropertyValue("inputConverter"));
assertSame(converter, dfa.getPropertyValue("outputConverter"));
assertSame(serializer, dfa.getPropertyValue("serializer"));
assertSame(deserializer, dfa.getPropertyValue("deserializer"));
assertEquals(true, dfa.getPropertyValue("soTcpNoDelay"));
assertEquals(true, dfa.getPropertyValue("singleUse"));
assertSame(taskExecutor, dfa.getPropertyValue("taskExecutor"));
@@ -283,8 +288,8 @@ public class ParserUnitTests {
assertEquals(1236, server1.getSoTimeout());
assertEquals(12, server1.getSoTrafficClass());
DirectFieldAccessor dfa = new DirectFieldAccessor(server1);
assertSame(converter, dfa.getPropertyValue("inputConverter"));
assertSame(converter, dfa.getPropertyValue("outputConverter"));
assertSame(serializer, dfa.getPropertyValue("serializer"));
assertSame(deserializer, dfa.getPropertyValue("deserializer"));
assertEquals(true, dfa.getPropertyValue("soTcpNoDelay"));
assertEquals(true, dfa.getPropertyValue("singleUse"));
assertSame(taskExecutor, dfa.getPropertyValue("taskExecutor"));
@@ -304,8 +309,8 @@ public class ParserUnitTests {
assertEquals(1236, client1.getSoTimeout());
assertEquals(12, client1.getSoTrafficClass());
DirectFieldAccessor dfa = new DirectFieldAccessor(client1);
assertSame(converter, dfa.getPropertyValue("inputConverter"));
assertSame(converter, dfa.getPropertyValue("outputConverter"));
assertSame(serializer, dfa.getPropertyValue("serializer"));
assertSame(deserializer, dfa.getPropertyValue("deserializer"));
assertEquals(true, dfa.getPropertyValue("soTcpNoDelay"));
assertEquals(true, dfa.getPropertyValue("singleUse"));
assertSame(taskExecutor, dfa.getPropertyValue("taskExecutor"));
@@ -323,8 +328,8 @@ public class ParserUnitTests {
assertEquals(1236, server1.getSoTimeout());
assertEquals(12, server1.getSoTrafficClass());
DirectFieldAccessor dfa = new DirectFieldAccessor(server1);
assertSame(converter, dfa.getPropertyValue("inputConverter"));
assertSame(converter, dfa.getPropertyValue("outputConverter"));
assertSame(serializer, dfa.getPropertyValue("serializer"));
assertSame(deserializer, dfa.getPropertyValue("deserializer"));
assertEquals(true, dfa.getPropertyValue("soTcpNoDelay"));
assertEquals(true, dfa.getPropertyValue("singleUse"));
assertSame(taskExecutor, dfa.getPropertyValue("taskExecutor"));

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

@@ -7,10 +7,11 @@
http://www.springframework.org/schema/integration/ip http://www.springframework.org/schema/integration/ip/spring-integration-ip.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd">
<bean id="tcpIpUtils" class="org.springframework.integration.ip.util.SocketUtils" />
<bean id="serializer" class="org.springframework.commons.serializer.java.JavaStreamingConverter" />
<bean id="serializer" class="org.springframework.commons.serializer.DefaultSerializer" />
<bean id="deserializer" class="org.springframework.commons.serializer.DefaultDeserializer" />
<bean id="helloWorldInterceptors" class="org.springframework.integration.ip.tcp.connection.TcpConnectionInterceptorFactoryChain">
<property name="interceptors">
@@ -27,8 +28,8 @@
<int-ip:tcp-connection-factory id="server"
type="server"
port="#{tcpIpUtils.findAvailableServerSocket(10100)}"
input-converter="serializer"
output-converter="serializer"
serializer="serializer"
deserializer="deserializer"
using-nio="true"
single-use="true"
interceptor-factory-chain="helloWorldInterceptors"
@@ -41,8 +42,8 @@
single-use="true"
so-timeout="100000"
using-nio="true"
input-converter="serializer"
output-converter="serializer"
serializer="serializer"
deserializer="deserializer"
interceptor-factory-chain="helloWorldInterceptors"
/>

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

@@ -11,13 +11,15 @@
<bean id="tcpIpUtils" class="org.springframework.integration.ip.util.SocketUtils" />
<bean id="serializer" class="org.springframework.commons.serializer.java.JavaStreamingConverter" />
<bean id="serializer" class="org.springframework.commons.serializer.DefaultSerializer" />
<bean id="deserializer" class="org.springframework.commons.serializer.DefaultDeserializer" />
<int-ip:tcp-connection-factory id="server"
type="server"
port="#{tcpIpUtils.findAvailableServerSocket(10200)}"
input-converter="serializer"
output-converter="serializer"
serializer="serializer"
deserializer="deserializer"
using-nio="true"
single-use="true"
/>
@@ -28,8 +30,8 @@
port="#{server.port}"
single-use="true"
so-timeout="10000"
input-converter="serializer"
output-converter="serializer"
serializer="serializer"
deserializer="deserializer"
/>
<int:channel id="input" />

1
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/SharedConnectionTests.java
View File

@@ -16,7 +16,6 @@
package org.springframework.integration.ip.tcp;
import static junit.framework.Assert.assertTrue;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.fail;

21
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java
View File

@@ -38,7 +38,8 @@ import javax.net.ServerSocketFactory;
import org.junit.Test;
import org.springframework.commons.serializer.java.JavaStreamingConverter;
import org.springframework.commons.serializer.DefaultDeserializer;
import org.springframework.commons.serializer.DefaultSerializer;
import org.springframework.integration.Message;
import org.springframework.integration.MessageTimeoutException;
import org.springframework.integration.channel.QueueChannel;
@@ -50,7 +51,6 @@ import org.springframework.integration.support.MessageBuilder;
/**
* @author Gary Russell
* @since 2.0
*
*/
public class TcpOutboundGatewayTests {
@@ -69,7 +69,7 @@ public class TcpOutboundGatewayTests {
while (true) {
Socket socket = server.accept();
ObjectInputStream ois = new ObjectInputStream(socket.getInputStream());
Object in = ois.readObject();
ois.readObject();
ObjectOutputStream oos = new ObjectOutputStream(socket.getOutputStream());
oos.writeObject("Reply" + (i++));
}
@@ -80,9 +80,8 @@ public class TcpOutboundGatewayTests {
}
}
});
JavaStreamingConverter converter = new JavaStreamingConverter();
ccf.setInputConverter(converter);
ccf.setOutputConverter(converter);
ccf.setSerializer(new DefaultSerializer());
ccf.setDeserializer(new DefaultDeserializer());
ccf.setSoTimeout(10000);
ccf.setSingleUse(true);
ccf.setPoolSize(10);
@@ -135,9 +134,8 @@ public class TcpOutboundGatewayTests {
}
});
AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port);
JavaStreamingConverter converter = new JavaStreamingConverter();
ccf.setInputConverter(converter);
ccf.setOutputConverter(converter);
ccf.setSerializer(new DefaultSerializer());
ccf.setDeserializer(new DefaultDeserializer());
ccf.setSoTimeout(10000);
ccf.setSingleUse(false);
ccf.start();
@@ -188,9 +186,8 @@ public class TcpOutboundGatewayTests {
}
});
AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port);
JavaStreamingConverter converter = new JavaStreamingConverter();
ccf.setInputConverter(converter);
ccf.setOutputConverter(converter);
ccf.setSerializer(new DefaultSerializer());
ccf.setDeserializer(new DefaultDeserializer());
ccf.setSoTimeout(10000);
ccf.setSingleUse(false);
ccf.start();

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

@@ -13,6 +13,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.ip.tcp;
import static org.junit.Assert.assertEquals;
@@ -35,7 +36,9 @@ import java.util.concurrent.Executors;
import javax.net.SocketFactory;
import org.junit.Test;
import org.springframework.commons.serializer.java.JavaStreamingConverter;
import org.springframework.commons.serializer.DefaultDeserializer;
import org.springframework.commons.serializer.DefaultSerializer;
import org.springframework.integration.Message;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory;
@@ -49,7 +52,6 @@ import org.springframework.integration.ip.util.SocketUtils;
/**
* @author Gary Russell
*
*/
public class TcpReceivingChannelAdapterTests {
@@ -58,8 +60,8 @@ public class TcpReceivingChannelAdapterTests {
final int port = SocketUtils.findAvailableServerSocket();
AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(port);
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
scf.setInputConverter(converter);
scf.setOutputConverter(converter);
scf.setSerializer(converter);
scf.setDeserializer(converter);
TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter();
adapter.setConnectionFactory(scf);
scf.start();
@@ -88,8 +90,8 @@ public class TcpReceivingChannelAdapterTests {
final int port = SocketUtils.findAvailableServerSocket();
TcpNioServerConnectionFactory scf = new TcpNioServerConnectionFactory(port);
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
scf.setInputConverter(converter);
scf.setOutputConverter(converter);
scf.setSerializer(converter);
scf.setDeserializer(converter);
scf.setSoTimeout(5000);
TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter();
adapter.setConnectionFactory(scf);
@@ -123,8 +125,8 @@ public class TcpReceivingChannelAdapterTests {
final int port = SocketUtils.findAvailableServerSocket();
AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(port);
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
scf.setInputConverter(converter);
scf.setOutputConverter(converter);
scf.setSerializer(converter);
scf.setDeserializer(converter);
TcpSendingMessageHandler handler = new TcpSendingMessageHandler();
handler.setConnectionFactory(scf);
TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter();
@@ -161,8 +163,8 @@ public class TcpReceivingChannelAdapterTests {
final int port = SocketUtils.findAvailableServerSocket();
TcpNioServerConnectionFactory scf = new TcpNioServerConnectionFactory(port);
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
scf.setInputConverter(converter);
scf.setOutputConverter(converter);
scf.setSerializer(converter);
scf.setDeserializer(converter);
TcpSendingMessageHandler handler = new TcpSendingMessageHandler();
handler.setConnectionFactory(scf);
TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter();
@@ -199,8 +201,8 @@ public class TcpReceivingChannelAdapterTests {
final int port = SocketUtils.findAvailableServerSocket();
AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(port);
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
scf.setInputConverter(converter);
scf.setOutputConverter(converter);
scf.setSerializer(converter);
scf.setDeserializer(converter);
scf.setSingleUse(true);
TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter();
adapter.setConnectionFactory(scf);
@@ -235,8 +237,8 @@ public class TcpReceivingChannelAdapterTests {
final int port = SocketUtils.findAvailableServerSocket();
TcpNioServerConnectionFactory scf = new TcpNioServerConnectionFactory(port);
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
scf.setInputConverter(converter);
scf.setOutputConverter(converter);
scf.setSerializer(converter);
scf.setDeserializer(converter);
scf.setSingleUse(true);
TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter();
adapter.setConnectionFactory(scf);
@@ -281,8 +283,8 @@ public class TcpReceivingChannelAdapterTests {
final int port = SocketUtils.findAvailableServerSocket();
AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(port);
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
scf.setInputConverter(converter);
scf.setOutputConverter(converter);
scf.setSerializer(converter);
scf.setDeserializer(converter);
scf.setSingleUse(true);
TcpSendingMessageHandler handler = new TcpSendingMessageHandler();
handler.setConnectionFactory(scf);
@@ -322,8 +324,8 @@ public class TcpReceivingChannelAdapterTests {
final int port = SocketUtils.findAvailableServerSocket();
TcpNioServerConnectionFactory scf = new TcpNioServerConnectionFactory(port);
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
scf.setInputConverter(converter);
scf.setOutputConverter(converter);
scf.setSerializer(converter);
scf.setDeserializer(converter);
scf.setSingleUse(true);
TcpSendingMessageHandler handler = new TcpSendingMessageHandler();
handler.setConnectionFactory(scf);
@@ -363,8 +365,8 @@ public class TcpReceivingChannelAdapterTests {
final int port = SocketUtils.findAvailableServerSocket();
TcpNioServerConnectionFactory scf = new TcpNioServerConnectionFactory(port);
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
scf.setInputConverter(converter);
scf.setOutputConverter(converter);
scf.setSerializer(converter);
scf.setDeserializer(converter);
scf.setSingleUse(true);
scf.setPoolSize(100);
TcpSendingMessageHandler handler = new TcpSendingMessageHandler();
@@ -444,11 +446,9 @@ public class TcpReceivingChannelAdapterTests {
singleSharedInterceptorsGuts(port, scf);
}
private void interceptorsGuts(final int port,
AbstractServerConnectionFactory scf) throws Exception {
JavaStreamingConverter converter = new JavaStreamingConverter();
scf.setInputConverter(converter);
scf.setOutputConverter(converter);
private void interceptorsGuts(final int port, AbstractServerConnectionFactory scf) throws Exception {
scf.setSerializer(new DefaultSerializer());
scf.setDeserializer(new DefaultDeserializer());
scf.setSingleUse(false);
TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter();
adapter.setConnectionFactory(scf);
@@ -487,11 +487,9 @@ public class TcpReceivingChannelAdapterTests {
assertTrue(results.contains("Test2"));
}
private void singleNoOutboundInterceptorsGuts(final int port,
AbstractServerConnectionFactory scf) throws Exception {
JavaStreamingConverter converter = new JavaStreamingConverter();
scf.setInputConverter(converter);
scf.setOutputConverter(converter);
private void singleNoOutboundInterceptorsGuts(final int port, AbstractServerConnectionFactory scf) throws Exception {
scf.setSerializer(new DefaultSerializer());
scf.setDeserializer(new DefaultDeserializer());
scf.setSingleUse(true);
scf.setSoTimeout(10000);
TcpConnectionInterceptorFactoryChain fc = new TcpConnectionInterceptorFactoryChain();
@@ -536,11 +534,9 @@ public class TcpReceivingChannelAdapterTests {
assertTrue(results.contains("Test2"));
}
private void singleSharedInterceptorsGuts(final int port,
AbstractServerConnectionFactory scf) throws Exception {
JavaStreamingConverter converter = new JavaStreamingConverter();
scf.setInputConverter(converter);
scf.setOutputConverter(converter);
private void singleSharedInterceptorsGuts(final int port, AbstractServerConnectionFactory scf) throws Exception {
scf.setSerializer(new DefaultSerializer());
scf.setDeserializer(new DefaultDeserializer());
scf.setSingleUse(true);
scf.setSoTimeout(60000);
TcpConnectionInterceptorFactoryChain fc = new TcpConnectionInterceptorFactoryChain();
@@ -588,7 +584,5 @@ public class TcpReceivingChannelAdapterTests {
assertEquals("Test1", new ObjectInputStream(socket1.getInputStream()).readObject());
assertEquals("Test2", new ObjectInputStream(socket2.getInputStream()).readObject());
}
}

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

@@ -13,6 +13,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.ip.tcp;
import static org.junit.Assert.assertEquals;
@@ -38,7 +39,9 @@ import java.util.concurrent.atomic.AtomicBoolean;
import javax.net.ServerSocketFactory;
import org.junit.Test;
import org.springframework.commons.serializer.java.JavaStreamingConverter;
import org.springframework.commons.serializer.DefaultDeserializer;
import org.springframework.commons.serializer.DefaultSerializer;
import org.springframework.integration.Message;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory;
@@ -53,11 +56,9 @@ import org.springframework.integration.ip.tcp.converter.ByteArrayStxEtxConverter
import org.springframework.integration.ip.util.SocketUtils;
import org.springframework.integration.support.MessageBuilder;
/**
* @author Gary Russell
* @since 2.0
*
*/
public class TcpSendingMessageHandlerTests {
@@ -94,8 +95,8 @@ public class TcpSendingMessageHandlerTests {
});
AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port);
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
ccf.setInputConverter(converter);
ccf.setOutputConverter(converter);
ccf.setSerializer(converter);
ccf.setDeserializer(converter);
ccf.setSoTimeout(10000);
TcpSendingMessageHandler handler = new TcpSendingMessageHandler();
handler.setConnectionFactory(ccf);
@@ -142,8 +143,8 @@ public class TcpSendingMessageHandlerTests {
});
AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port);
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
ccf.setInputConverter(converter);
ccf.setOutputConverter(converter);
ccf.setSerializer(converter);
ccf.setDeserializer(converter);
ccf.setSoTimeout(10000);
ccf.start();
TcpSendingMessageHandler handler = new TcpSendingMessageHandler();
@@ -194,8 +195,8 @@ public class TcpSendingMessageHandlerTests {
});
AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port);
ByteArrayStxEtxConverter converter = new ByteArrayStxEtxConverter();
ccf.setInputConverter(converter);
ccf.setOutputConverter(converter);
ccf.setSerializer(converter);
ccf.setDeserializer(converter);
ccf.setSoTimeout(10000);
ccf.start();
TcpSendingMessageHandler handler = new TcpSendingMessageHandler();
@@ -243,8 +244,8 @@ public class TcpSendingMessageHandlerTests {
});
AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port);
ByteArrayStxEtxConverter converter = new ByteArrayStxEtxConverter();
ccf.setInputConverter(converter);
ccf.setOutputConverter(converter);
ccf.setSerializer(converter);
ccf.setDeserializer(converter);
ccf.setSoTimeout(10000);
ccf.start();
TcpSendingMessageHandler handler = new TcpSendingMessageHandler();
@@ -298,8 +299,8 @@ public class TcpSendingMessageHandlerTests {
});
AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port);
ByteArrayLengthHeaderConverter converter = new ByteArrayLengthHeaderConverter();
ccf.setInputConverter(converter);
ccf.setOutputConverter(converter);
ccf.setSerializer(converter);
ccf.setDeserializer(converter);
ccf.setSoTimeout(10000);
ccf.start();
TcpSendingMessageHandler handler = new TcpSendingMessageHandler();
@@ -350,8 +351,8 @@ public class TcpSendingMessageHandlerTests {
});
AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port);
ByteArrayLengthHeaderConverter converter = new ByteArrayLengthHeaderConverter();
ccf.setInputConverter(converter);
ccf.setOutputConverter(converter);
ccf.setSerializer(converter);
ccf.setDeserializer(converter);
ccf.setSoTimeout(10000);
ccf.start();
TcpSendingMessageHandler handler = new TcpSendingMessageHandler();
@@ -401,9 +402,8 @@ public class TcpSendingMessageHandlerTests {
}
});
AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port);
JavaStreamingConverter converter = new JavaStreamingConverter();
ccf.setInputConverter(converter);
ccf.setOutputConverter(converter);
ccf.setSerializer(new DefaultSerializer());
ccf.setDeserializer(new DefaultDeserializer());
ccf.setSoTimeout(10000);
ccf.start();
TcpSendingMessageHandler handler = new TcpSendingMessageHandler();
@@ -450,9 +450,8 @@ public class TcpSendingMessageHandlerTests {
}
});
AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port);
JavaStreamingConverter converter = new JavaStreamingConverter();
ccf.setInputConverter(converter);
ccf.setOutputConverter(converter);
ccf.setSerializer(new DefaultSerializer());
ccf.setDeserializer(new DefaultDeserializer());
ccf.setSoTimeout(10000);
ccf.start();
TcpSendingMessageHandler handler = new TcpSendingMessageHandler();
@@ -503,8 +502,8 @@ public class TcpSendingMessageHandlerTests {
});
AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port);
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
ccf.setInputConverter(converter);
ccf.setOutputConverter(converter);
ccf.setSerializer(converter);
ccf.setDeserializer(converter);
ccf.setSoTimeout(10000);
ccf.start();
ccf.setSingleUse(true);
@@ -544,8 +543,8 @@ public class TcpSendingMessageHandlerTests {
});
AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port);
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
ccf.setInputConverter(converter);
ccf.setOutputConverter(converter);
ccf.setSerializer(converter);
ccf.setDeserializer(converter);
ccf.setSoTimeout(10000);
ccf.start();
ccf.setSingleUse(true);
@@ -587,8 +586,8 @@ public class TcpSendingMessageHandlerTests {
});
AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port);
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
ccf.setInputConverter(converter);
ccf.setOutputConverter(converter);
ccf.setSerializer(converter);
ccf.setDeserializer(converter);
ccf.setSoTimeout(10000);
ccf.start();
ccf.setSingleUse(true);
@@ -642,8 +641,8 @@ public class TcpSendingMessageHandlerTests {
});
AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port);
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
ccf.setInputConverter(converter);
ccf.setOutputConverter(converter);
ccf.setSerializer(converter);
ccf.setDeserializer(converter);
ccf.setSoTimeout(10000);
ccf.start();
ccf.setSingleUse(true);
@@ -697,8 +696,8 @@ public class TcpSendingMessageHandlerTests {
});
AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port);
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
ccf.setInputConverter(converter);
ccf.setOutputConverter(converter);
ccf.setSerializer(converter);
ccf.setDeserializer(converter);
ccf.setSoTimeout(10000);
ccf.setSingleUse(true);
ccf.setTaskExecutor(Executors.newFixedThreadPool(100));
@@ -771,9 +770,8 @@ public class TcpSendingMessageHandlerTests {
}
});
AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port);
JavaStreamingConverter converter = new JavaStreamingConverter();
ccf.setInputConverter(converter);
ccf.setOutputConverter(converter);
ccf.setSerializer(new DefaultSerializer());
ccf.setDeserializer(new DefaultDeserializer());
ccf.setSoTimeout(10000);
TcpConnectionInterceptorFactoryChain fc = new TcpConnectionInterceptorFactoryChain();
fc.setInterceptors(new TcpConnectionInterceptorFactory[]
@@ -834,9 +832,8 @@ public class TcpSendingMessageHandlerTests {
}
});
AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port);
JavaStreamingConverter converter = new JavaStreamingConverter();
ccf.setInputConverter(converter);
ccf.setOutputConverter(converter);
ccf.setSerializer(new DefaultSerializer());
ccf.setDeserializer(new DefaultDeserializer());
ccf.setSoTimeout(10000);
TcpConnectionInterceptorFactoryChain fc = new TcpConnectionInterceptorFactoryChain();
fc.setInterceptors(new TcpConnectionInterceptorFactory[] {new HelloWorldInterceptorFactory()});
@@ -904,9 +901,8 @@ public class TcpSendingMessageHandlerTests {
}
});
AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port);
JavaStreamingConverter converter = new JavaStreamingConverter();
ccf.setInputConverter(converter);
ccf.setOutputConverter(converter);
ccf.setSerializer(new DefaultSerializer());
ccf.setDeserializer(new DefaultDeserializer());
ccf.setSoTimeout(10000);
TcpConnectionInterceptorFactoryChain fc = new TcpConnectionInterceptorFactoryChain();
fc.setInterceptors(new TcpConnectionInterceptorFactory[]
@@ -961,9 +957,8 @@ public class TcpSendingMessageHandlerTests {
}
});
AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port);
JavaStreamingConverter converter = new JavaStreamingConverter();
ccf.setInputConverter(converter);
ccf.setOutputConverter(converter);
ccf.setSerializer(new DefaultSerializer());
ccf.setDeserializer(new DefaultDeserializer());
ccf.setSoTimeout(10000);
TcpConnectionInterceptorFactoryChain fc = new TcpConnectionInterceptorFactoryChain();
fc.setInterceptors(new TcpConnectionInterceptorFactory[]

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

@@ -15,119 +15,119 @@
<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="javaConverter" class="org.springframework.commons.serializer.java.JavaStreamingConverter" />
<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)}"
input-converter="crLfConverter"
output-converter="crLfConverter"/>
serializer="crLfConverter"
deserializer="crLfConverter"/>
<ip:tcp-connection-factory id="stxEtxServer"
type="server"
port="#{tcpIpUtils.findAvailableServerSocket(7100)}"
input-converter="stxEtxConverter"
output-converter="stxEtxConverter"/>
serializer="stxEtxConverter"
deserializer="stxEtxConverter"/>
<ip:tcp-connection-factory id="lengthHeaderServer"
type="server"
port="#{tcpIpUtils.findAvailableServerSocket(7200)}"
input-converter="lengthHeaderConverter"
output-converter="lengthHeaderConverter"/>
serializer="lengthHeaderConverter"
deserializer="lengthHeaderConverter"/>
<ip:tcp-connection-factory id="javaSerialServer"
type="server"
port="#{tcpIpUtils.findAvailableServerSocket(7300)}"
input-converter="javaConverter"
output-converter="javaConverter"/>
serializer="javaSerializer"
deserializer="javaDeserializer"/>
<ip:tcp-connection-factory id="crLfClient"
type="client"
host="localhost"
port="#{crLfServer.port}"
input-converter="crLfConverter"
output-converter="crLfConverter"/>
serializer="crLfConverter"
deserializer="crLfConverter"/>
<ip:tcp-connection-factory id="stxEtxClient"
type="client"
host="localhost"
port="#{stxEtxServer.port}"
input-converter="stxEtxConverter"
output-converter="stxEtxConverter"/>
serializer="stxEtxConverter"
deserializer="stxEtxConverter"/>
<ip:tcp-connection-factory id="lengthHeaderClient"
type="client"
host="localhost"
port="#{lengthHeaderServer.port}"
input-converter="lengthHeaderConverter"
output-converter="lengthHeaderConverter"/>
serializer="lengthHeaderConverter"
deserializer="lengthHeaderConverter"/>
<ip:tcp-connection-factory id="javaSerialClient"
type="client"
host="localhost"
port="#{javaSerialServer.port}"
input-converter="javaConverter"
output-converter="javaConverter"/>
serializer="javaSerializer"
deserializer="javaDeserializer"/>
<ip:tcp-connection-factory id="crLfServerNio"
type="server"
port="#{tcpIpUtils.findAvailableServerSocket(7400)}"
input-converter="crLfConverter"
output-converter="crLfConverter"
serializer="crLfConverter"
deserializer="crLfConverter"
using-nio="true"/>
<ip:tcp-connection-factory id="stxEtxServerNio"
type="server"
port="#{tcpIpUtils.findAvailableServerSocket(7500)}"
input-converter="stxEtxConverter"
output-converter="stxEtxConverter"
serializer="stxEtxConverter"
deserializer="stxEtxConverter"
using-nio="true"/>
<ip:tcp-connection-factory id="lengthHeaderServerNio"
type="server"
port="#{tcpIpUtils.findAvailableServerSocket(7600)}"
input-converter="lengthHeaderConverter"
output-converter="lengthHeaderConverter"
serializer="lengthHeaderConverter"
deserializer="lengthHeaderConverter"
using-nio="true"/>
<ip:tcp-connection-factory id="javaSerialServerNio"
type="server"
port="#{tcpIpUtils.findAvailableServerSocket(7700)}"
input-converter="javaConverter"
output-converter="javaConverter"
serializer="javaSerializer"
deserializer="javaDeserializer"
using-nio="true"/>
<ip:tcp-connection-factory id="crLfClientNio"
type="client"
host="localhost"
port="#{crLfServer.port}"
input-converter="crLfConverter"
output-converter="crLfConverter"
serializer="crLfConverter"
deserializer="crLfConverter"
using-nio="true"/>
<ip:tcp-connection-factory id="stxEtxClientNio"
type="client"
host="localhost"
port="#{stxEtxServer.port}"
input-converter="stxEtxConverter"
output-converter="stxEtxConverter"
serializer="stxEtxConverter"
deserializer="stxEtxConverter"
using-nio="true"/>
<ip:tcp-connection-factory id="lengthHeaderClientNio"
type="client"
host="localhost"
port="#{lengthHeaderServer.port}"
input-converter="lengthHeaderConverter"
output-converter="lengthHeaderConverter"
serializer="lengthHeaderConverter"
deserializer="lengthHeaderConverter"
using-nio="true"/>
<ip:tcp-connection-factory id="javaSerialClientNio"
type="client"
host="localhost"
port="#{javaSerialServer.port}"
input-converter="javaConverter"
output-converter="javaConverter"
serializer="javaSerializer"
deserializer="javaDeserializer"
using-nio="true"/>
</beans>

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

@@ -13,6 +13,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.ip.tcp.connection;
import static org.junit.Assert.assertEquals;
@@ -31,6 +32,7 @@ import java.util.concurrent.TimeUnit;
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;
@@ -40,7 +42,7 @@ import org.springframework.integration.ip.util.SocketUtils;
/**
* @author Gary Russell
*
* @since 2.0
*/
public class TcpNioConnectionReadTests {
@@ -54,8 +56,8 @@ public class TcpNioConnectionReadTests {
private AbstractServerConnectionFactory getConnectionFactory(int port,
AbstractByteArrayStreamingConverter converter, TcpListener listener, TcpSender sender) throws Exception {
AbstractServerConnectionFactory scf = new TcpNioServerConnectionFactory(port);
scf.setInputConverter(converter);
scf.setOutputConverter(converter);
scf.setSerializer(converter);
scf.setDeserializer(converter);
scf.registerListener(listener);
if (sender != null) {
scf.registerSender(sender);

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

@@ -13,6 +13,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.ip.tcp.connection;
import static org.junit.Assert.assertEquals;
@@ -26,6 +27,7 @@ import java.nio.ByteBuffer;
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;
@@ -35,15 +37,15 @@ import org.springframework.integration.support.MessageBuilder;
/**
* @author Gary Russell
*
* @since 2.0
*/
public class TcpNioConnectionWriteTests {
private AbstractConnectionFactory getClientConnectionFactory(boolean direct,
final int port, AbstractByteArrayStreamingConverter converter) {
TcpNioClientConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port);
ccf.setInputConverter(converter);
ccf.setOutputConverter(converter);
ccf.setSerializer(converter);
ccf.setDeserializer(converter);
ccf.setSoTimeout(10000);
ccf.setUsingDirectBuffers(direct);
ccf.start();

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

@@ -13,6 +13,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.ip.tcp.converter;
import static org.junit.Assert.assertEquals;
@@ -25,14 +26,15 @@ import java.net.Socket;
import javax.net.ServerSocketFactory;
import org.junit.Test;
import org.springframework.commons.serializer.java.JavaStreamingConverter;
import org.springframework.commons.serializer.DefaultDeserializer;
import org.springframework.integration.ip.util.SocketUtils;
/**
* @author Gary Russell
*
* @since 2.0
*/
public class InputConverterTests {
public class DeserializationTests {
@Test
public void testReadLength() throws Exception {
@@ -43,10 +45,10 @@ public class InputConverterTests {
Socket socket = server.accept();
socket.setSoTimeout(5000);
ByteArrayLengthHeaderConverter converter = new ByteArrayLengthHeaderConverter();
byte[] out = converter.convert(socket.getInputStream());
byte[] out = converter.deserialize(socket.getInputStream());
assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING,
new String(out));
out = converter.convert(socket.getInputStream());
out = converter.deserialize(socket.getInputStream());
assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING,
new String(out));
server.close();
@@ -61,10 +63,10 @@ public class InputConverterTests {
Socket socket = server.accept();
socket.setSoTimeout(5000);
ByteArrayStxEtxConverter converter = new ByteArrayStxEtxConverter();
byte[] out = converter.convert(socket.getInputStream());
byte[] out = converter.deserialize(socket.getInputStream());
assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING,
new String(out));
out = converter.convert(socket.getInputStream());
out = converter.deserialize(socket.getInputStream());
assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING,
new String(out));
server.close();
@@ -79,10 +81,10 @@ public class InputConverterTests {
Socket socket = server.accept();
socket.setSoTimeout(5000);
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
byte[] out = converter.convert(socket.getInputStream());
byte[] out = converter.deserialize(socket.getInputStream());
assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING,
new String(out));
out = converter.convert(socket.getInputStream());
out = converter.deserialize(socket.getInputStream());
assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING,
new String(out));
server.close();
@@ -96,10 +98,10 @@ public class InputConverterTests {
SocketUtils.testSendSerialized(port);
Socket socket = server.accept();
socket.setSoTimeout(5000);
JavaStreamingConverter converter = new JavaStreamingConverter();
Object out = converter.convert(socket.getInputStream());
DefaultDeserializer deserializer = new DefaultDeserializer();
Object out = deserializer.deserialize(socket.getInputStream());
assertEquals("Data", SocketUtils.TEST_STRING, out);
out = converter.convert(socket.getInputStream());
out = deserializer.deserialize(socket.getInputStream());
assertEquals("Data", SocketUtils.TEST_STRING, out);
server.close();
}
@@ -114,7 +116,7 @@ public class InputConverterTests {
socket.setSoTimeout(5000);
ByteArrayLengthHeaderConverter converter = new ByteArrayLengthHeaderConverter();
try {
converter.convert(socket.getInputStream());
converter.deserialize(socket.getInputStream());
fail("Expected message length exceeded exception");
} catch (IOException e) {
if (!e.getMessage().startsWith("Message length")) {
@@ -135,7 +137,7 @@ public class InputConverterTests {
socket.setSoTimeout(500);
ByteArrayStxEtxConverter converter = new ByteArrayStxEtxConverter();
try {
converter.convert(socket.getInputStream());
converter.deserialize(socket.getInputStream());
fail("Expected timeout exception");
} catch (IOException e) {
if (!e.getMessage().startsWith("Read timed out")) {
@@ -157,7 +159,7 @@ public class InputConverterTests {
ByteArrayStxEtxConverter converter = new ByteArrayStxEtxConverter();
converter.setMaxMessageSize(1024);
try {
converter.convert(socket.getInputStream());
converter.deserialize(socket.getInputStream());
fail("Expected message length exceeded exception");
} catch (IOException e) {
if (!e.getMessage().startsWith("ETX not found")) {
@@ -178,7 +180,7 @@ public class InputConverterTests {
socket.setSoTimeout(500);
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
try {
converter.convert(socket.getInputStream());
converter.deserialize(socket.getInputStream());
fail("Expected timout exception");
} catch (IOException e) {
if (!e.getMessage().startsWith("Read timed out")) {
@@ -200,7 +202,7 @@ public class InputConverterTests {
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
converter.setMaxMessageSize(1024);
try {
converter.convert(socket.getInputStream());
converter.deserialize(socket.getInputStream());
fail("Expected message length exceeded exception");
} catch (IOException e) {
if (!e.getMessage().startsWith("CRLF not found")) {
@@ -210,5 +212,5 @@ public class InputConverterTests {
}
server.close();
}
}

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

@@ -13,6 +13,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.ip.tcp.converter;
import static org.junit.Assert.assertEquals;
@@ -28,14 +29,15 @@ import javax.net.ServerSocketFactory;
import javax.net.SocketFactory;
import org.junit.Test;
import org.springframework.commons.serializer.java.JavaStreamingConverter;
import org.springframework.commons.serializer.DefaultSerializer;
import org.springframework.integration.ip.util.SocketUtils;
/**
* @author Gary Russell
*
* @since 2.0
*/
public class OutputConverterTests {
public class SerializationTests {
@Test
public void testWriteLengthHeader() throws Exception {
@@ -50,7 +52,7 @@ public class OutputConverterTests {
ByteBuffer buffer = ByteBuffer.allocate(testString.length());
buffer.put(testString.getBytes());
ByteArrayLengthHeaderConverter converter = new ByteArrayLengthHeaderConverter();
converter.convert(buffer.array(), socket.getOutputStream());
converter.serialize(buffer.array(), socket.getOutputStream());
Thread.sleep(1000000000L);
} catch (Exception e) {
e.printStackTrace();
@@ -83,7 +85,7 @@ public class OutputConverterTests {
ByteBuffer buffer = ByteBuffer.allocate(testString.length());
buffer.put(testString.getBytes());
ByteArrayStxEtxConverter converter = new ByteArrayStxEtxConverter();
converter.convert(buffer.array(), socket.getOutputStream());
converter.serialize(buffer.array(), socket.getOutputStream());
Thread.sleep(1000000000L);
} catch (Exception e) {
e.printStackTrace();
@@ -116,7 +118,7 @@ public class OutputConverterTests {
ByteBuffer buffer = ByteBuffer.allocate(testString.length());
buffer.put(testString.getBytes());
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
converter.convert(buffer.array(), socket.getOutputStream());
converter.serialize(buffer.array(), socket.getOutputStream());
Thread.sleep(1000000000L);
} catch (Exception e) {
e.printStackTrace();
@@ -146,9 +148,9 @@ public class OutputConverterTests {
public void run() {
try {
Socket socket = SocketFactory.getDefault().createSocket("localhost", port);
JavaStreamingConverter converter = new JavaStreamingConverter();
converter.convert(testString, socket.getOutputStream());
converter.convert(testString, socket.getOutputStream());
DefaultSerializer serializer = new DefaultSerializer();
serializer.serialize(testString, socket.getOutputStream());
serializer.serialize(testString, socket.getOutputStream());
Thread.sleep(1000000000L);
} catch (Exception e) {
e.printStackTrace();