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

INT-1486 polish; rename byte array 'serializers' and all test usage

Browse Source
This commit is contained in:
Gary Russell
2010-09-28 22:25:04 -04:00
parent 4ae4b5d4d7
commit b339157a34
23 changed files with 172 additions and 951 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}.

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

@@ -1,67 +0,0 @@
/*
* Copyright 2002-2010 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.ip.tcp.converter;
import java.io.IOException;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.commons.serializer.Deserializer;
import org.springframework.commons.serializer.Serializer;
/**
* Base class for streaming converters that convert to/from a byte array.
*
* @author Gary Russell
* @since 2.0
*
*/
public abstract class AbstractByteArrayStreamingConverter implements
Serializer<byte[]>,
Deserializer<byte[]> {
protected int maxMessageSize = 2048;
protected Log logger = LogFactory.getLog(this.getClass());
/**
* The maximum supported message size for this converter.
* Default 2048.
* @return The max message size.
*/
public int getMaxMessageSize() {
return maxMessageSize;
}
/**
* The maximum supported message size for this converter.
* Default 2048.
* @param maxMessageSize The max message size.
*/
public void setMaxMessageSize(int maxMessageSize) {
this.maxMessageSize = maxMessageSize;
}
protected void checkClosure(int bite) throws IOException {
if (bite < 0) {
logger.debug("Socket closed");
throw new IOException("Socket closed");
}
}
}

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

@@ -1,75 +0,0 @@
/*
* Copyright 2002-2010 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.ip.tcp.converter;
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
* (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 {
/**
* Converts 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).
*/
public byte[] deserialize(InputStream inputStream) throws IOException {
byte[] buffer = new byte[this.maxMessageSize];
int n = 0;
int bite;
if (logger.isDebugEnabled())
logger.debug("Available to read:" + inputStream.available());
while (true) {
bite = inputStream.read();
// logger.debug("Read:" + (char) bite);
if (bite < 0 && n == 0) {
throw new SoftEndOfStreamException("Stream closed between payloads");
}
checkClosure(bite);
if (n > 0 && bite == '\n' && buffer[n-1] == '\r')
break;
buffer[n++] = (byte) bite;
if (n >= this.maxMessageSize) {
throw new IOException("CRLF not found before max message length: "
+ this.maxMessageSize);
}
};
byte[] assembledData = new byte[n-1];
System.arraycopy(buffer, 0, assembledData, 0, n-1);
return assembledData;
}
/**
* Writes the byte[] to the stream and appends \r\n.
*/
public void serialize(byte[] bytes, OutputStream outputStream) throws IOException {
outputStream.write(bytes);
outputStream.write('\r');
outputStream.write('\n');
outputStream.flush();
}
}

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

@@ -1,113 +0,0 @@
/*
* Copyright 2002-2010 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.ip.tcp.converter;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.ByteBuffer;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
/**
* Converts data in an InputStream to a byte[]; data is 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.\
* 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 {
private Log logger = LogFactory.getLog(this.getClass());
/**
* Reads a 4 byte length from the stream and then reads that length
* from the stream and returns the data in a byte[]. Throws an
* IOException if the length field exceeds the maxMessageSize.
* Throws a {@link SoftEndOfStreamException} if the stream
* is closed between messages.
*/
public byte[] deserialize(InputStream inputStream) throws IOException {
byte[] lengthPart = new byte[4];
int status = read(inputStream, lengthPart, true);
if (status < 0) {
throw new SoftEndOfStreamException("Stream closed between payloads");
}
int messageLength = ByteBuffer.wrap(lengthPart).getInt();
if (logger.isDebugEnabled()) {
logger.debug("Message length is " + messageLength);
}
if (messageLength > this.maxMessageSize) {
throw new IOException("Message length " + messageLength +
" exceeds max message length: " + this.maxMessageSize);
}
byte[] messagePart = new byte[messageLength];
read(inputStream, messagePart, false);
return messagePart;
}
/**
* Writes the byte[] to the output stream, preceded by a 4 byte
* length in network byte order (big endian).
*/
public void serialize(byte[] bytes, OutputStream outputStream) throws IOException {
ByteBuffer lengthPart = ByteBuffer.allocate(4);
lengthPart.putInt(bytes.length);
outputStream.write(lengthPart.array());
outputStream.write(bytes);
outputStream.flush();
}
/**
* Reads data from the socket and puts the data in buffer. Blocks until
* buffer is full or a socket timeout occurs.
* @param buffer
* @param header true if we are reading the header
* @return < 0 if socket closed and not in the middle of a message
* @throws IOException
*/
protected int read(InputStream inputStream, byte[] buffer, boolean header)
throws IOException {
int lengthRead = 0;
int needed = buffer.length;
while (lengthRead < needed) {
int len;
len = inputStream.read(buffer, lengthRead,
needed - lengthRead);
if (len < 0 && header && lengthRead == 0) {
return len;
}
if (len < 0) {
throw new IOException("Stream closed after " + lengthRead + " of " + needed);
}
lengthRead += len;
if (logger.isDebugEnabled()) {
logger.debug("Read " + len + " bytes, buffer is now at " +
lengthRead + " of " +
needed);
}
}
return 0;
}
}

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

@@ -1,80 +0,0 @@
/*
* Copyright 2002-2010 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.ip.tcp.converter;
import java.io.IOException;
import java.io.InputStream;
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
* 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;
*
* @author Gary Russell
* @since 2.0
*/
public class ByteArrayStxEtxConverter extends AbstractByteArrayStreamingConverter {
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
* 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
* being read).
*
*/
public byte[] deserialize(InputStream inputStream) throws IOException {
int bite = inputStream.read();
if (bite < 0) {
throw new SoftEndOfStreamException("Stream closed between payloads");
}
if (bite != STX)
throw new MessageMappingException("Expected STX to begin message");
byte[] buffer = new byte[this.maxMessageSize];
int n = 0;
while ((bite = inputStream.read()) != ETX) {
checkClosure(bite);
buffer[n++] = (byte) bite;
if (n >= this.maxMessageSize) {
throw new IOException("ETX not found before max message length: "
+ this.maxMessageSize);
}
}
byte[] assembledData = new byte[n];
System.arraycopy(buffer, 0, assembledData, 0, n);
return assembledData;
}
/**
* Writes the byte[] to the stream, prefixed by an ASCII STX character and
* terminated with an ASCII ETX character.
*/
public void serialize(byte[] bytes, OutputStream outputStream) throws IOException {
outputStream.write(STX);
outputStream.write(bytes);
outputStream.write(ETX);
outputStream.flush();
}
}

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

@@ -1,41 +0,0 @@
/*
* Copyright 2002-2010 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.ip.tcp.converter;
import java.io.IOException;
/**
* Used to communicate that a stream has closed, but between logical
* messages.
*
* @author Gary Russell
* @since 2.0
*
*/
public class SoftEndOfStreamException extends IOException {
private static final long serialVersionUID = 7309907445617226978L;
public SoftEndOfStreamException() {
super();
}
public SoftEndOfStreamException(String message) {
super(message);
}
}

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;

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

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

@@ -1,216 +0,0 @@
/*
* Copyright 2002-2010 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* 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;
import static org.junit.Assert.fail;
import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
import javax.net.ServerSocketFactory;
import org.junit.Test;
import org.springframework.commons.serializer.DefaultDeserializer;
import org.springframework.integration.ip.util.SocketUtils;
/**
* @author Gary Russell
* @since 2.0
*/
public class DeserializationTests {
@Test
public void testReadLength() throws Exception {
int port = SocketUtils.findAvailableServerSocket();
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
server.setSoTimeout(10000);
SocketUtils.testSendLength(port, null);
Socket socket = server.accept();
socket.setSoTimeout(5000);
ByteArrayLengthHeaderConverter converter = new ByteArrayLengthHeaderConverter();
byte[] out = converter.deserialize(socket.getInputStream());
assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING,
new String(out));
out = converter.deserialize(socket.getInputStream());
assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING,
new String(out));
server.close();
}
@Test
public void testReadStxEtx() throws Exception {
int port = SocketUtils.findAvailableServerSocket();
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
server.setSoTimeout(10000);
SocketUtils.testSendStxEtx(port, null);
Socket socket = server.accept();
socket.setSoTimeout(5000);
ByteArrayStxEtxConverter converter = new ByteArrayStxEtxConverter();
byte[] out = converter.deserialize(socket.getInputStream());
assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING,
new String(out));
out = converter.deserialize(socket.getInputStream());
assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING,
new String(out));
server.close();
}
@Test
public void testReadCrLf() throws Exception {
int port = SocketUtils.findAvailableServerSocket();
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
server.setSoTimeout(10000);
SocketUtils.testSendCrLf(port, null);
Socket socket = server.accept();
socket.setSoTimeout(5000);
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
byte[] out = converter.deserialize(socket.getInputStream());
assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING,
new String(out));
out = converter.deserialize(socket.getInputStream());
assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING,
new String(out));
server.close();
}
@Test
public void testReadSerialized() throws Exception {
int port = SocketUtils.findAvailableServerSocket();
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
server.setSoTimeout(10000);
SocketUtils.testSendSerialized(port);
Socket socket = server.accept();
socket.setSoTimeout(5000);
DefaultDeserializer deserializer = new DefaultDeserializer();
Object out = deserializer.deserialize(socket.getInputStream());
assertEquals("Data", SocketUtils.TEST_STRING, out);
out = deserializer.deserialize(socket.getInputStream());
assertEquals("Data", SocketUtils.TEST_STRING, out);
server.close();
}
@Test
public void testReadLengthOverflow() throws Exception {
int port = SocketUtils.findAvailableServerSocket();
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
server.setSoTimeout(10000);
SocketUtils.testSendLengthOverflow(port);
Socket socket = server.accept();
socket.setSoTimeout(5000);
ByteArrayLengthHeaderConverter converter = new ByteArrayLengthHeaderConverter();
try {
converter.deserialize(socket.getInputStream());
fail("Expected message length exceeded exception");
} catch (IOException e) {
if (!e.getMessage().startsWith("Message length")) {
e.printStackTrace();
fail("Unexpected IO Error:" + e.getMessage());
}
}
server.close();
}
@Test
public void testReadStxEtxTimeout() throws Exception {
int port = SocketUtils.findAvailableServerSocket();
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
server.setSoTimeout(10000);
SocketUtils.testSendStxEtxOverflow(port);
Socket socket = server.accept();
socket.setSoTimeout(500);
ByteArrayStxEtxConverter converter = new ByteArrayStxEtxConverter();
try {
converter.deserialize(socket.getInputStream());
fail("Expected timeout exception");
} catch (IOException e) {
if (!e.getMessage().startsWith("Read timed out")) {
e.printStackTrace();
fail("Unexpected IO Error:" + e.getMessage());
}
}
server.close();
}
@Test
public void testReadStxEtxOverflow() throws Exception {
int port = SocketUtils.findAvailableServerSocket();
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
server.setSoTimeout(10000);
SocketUtils.testSendStxEtxOverflow(port);
Socket socket = server.accept();
socket.setSoTimeout(5000);
ByteArrayStxEtxConverter converter = new ByteArrayStxEtxConverter();
converter.setMaxMessageSize(1024);
try {
converter.deserialize(socket.getInputStream());
fail("Expected message length exceeded exception");
} catch (IOException e) {
if (!e.getMessage().startsWith("ETX not found")) {
e.printStackTrace();
fail("Unexpected IO Error:" + e.getMessage());
}
}
server.close();
}
@Test
public void testReadCrLfTimeout() throws Exception {
int port = SocketUtils.findAvailableServerSocket();
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
server.setSoTimeout(10000);
SocketUtils.testSendCrLfOverflow(port);
Socket socket = server.accept();
socket.setSoTimeout(500);
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
try {
converter.deserialize(socket.getInputStream());
fail("Expected timout exception");
} catch (IOException e) {
if (!e.getMessage().startsWith("Read timed out")) {
e.printStackTrace();
fail("Unexpected IO Error:" + e.getMessage());
}
}
server.close();
}
@Test
public void testReadCrLfOverflow() throws Exception {
int port = SocketUtils.findAvailableServerSocket();
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
server.setSoTimeout(10000);
SocketUtils.testSendCrLfOverflow(port);
Socket socket = server.accept();
socket.setSoTimeout(5000);
ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter();
converter.setMaxMessageSize(1024);
try {
converter.deserialize(socket.getInputStream());
fail("Expected message length exceeded exception");
} catch (IOException e) {
if (!e.getMessage().startsWith("CRLF not found")) {
e.printStackTrace();
fail("Unexpected IO Error:" + e.getMessage());
}
}
server.close();
}
}

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

@@ -1,182 +0,0 @@
/*
* Copyright 2002-2010 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* 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;
import java.io.IOException;
import java.io.InputStream;
import java.io.ObjectInputStream;
import java.net.ServerSocket;
import java.net.Socket;
import java.nio.ByteBuffer;
import javax.net.ServerSocketFactory;
import javax.net.SocketFactory;
import org.junit.Test;
import org.springframework.commons.serializer.DefaultSerializer;
import org.springframework.integration.ip.util.SocketUtils;
/**
* @author Gary Russell
* @since 2.0
*/
public class SerializationTests {
@Test
public void testWriteLengthHeader() throws Exception {
final int port = SocketUtils.findAvailableServerSocket();
final String testString = "abcdef";
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
server.setSoTimeout(10000);
Thread t = new Thread(new Runnable() {
public void run() {
try {
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());
Thread.sleep(1000000000L);
} catch (Exception e) {
e.printStackTrace();
}
}
});
t.setDaemon(true);
t.start();
Socket socket = server.accept();
socket.setSoTimeout(5000);
InputStream is = socket.getInputStream();
byte[] buff = new byte[testString.length() + 4];
readFully(is, buff);
ByteBuffer buffer = ByteBuffer.wrap(buff);
assertEquals(testString.length(), buffer.getInt());
assertEquals(testString, new String(buff, 4, testString.length()));
server.close();
}
@Test
public void testWriteStxEtx() throws Exception {
final int port = SocketUtils.findAvailableServerSocket();
final String testString = "abcdef";
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
server.setSoTimeout(10000);
Thread t = new Thread(new Runnable() {
public void run() {
try {
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());
Thread.sleep(1000000000L);
} catch (Exception e) {
e.printStackTrace();
}
}
});
t.setDaemon(true);
t.start();
Socket socket = server.accept();
socket.setSoTimeout(5000);
InputStream is = socket.getInputStream();
byte[] buff = new byte[testString.length() + 2];
readFully(is, buff);
assertEquals(ByteArrayStxEtxConverter.STX, buff[0]);
assertEquals(testString, new String(buff, 1, testString.length()));
assertEquals(ByteArrayStxEtxConverter.ETX, buff[testString.length() + 1]);
server.close();
}
@Test
public void testWriteCrLf() throws Exception {
final int port = SocketUtils.findAvailableServerSocket();
final String testString = "abcdef";
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
server.setSoTimeout(10000);
Thread t = new Thread(new Runnable() {
public void run() {
try {
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());
Thread.sleep(1000000000L);
} catch (Exception e) {
e.printStackTrace();
}
}
});
t.setDaemon(true);
t.start();
Socket socket = server.accept();
socket.setSoTimeout(5000);
InputStream is = socket.getInputStream();
byte[] buff = new byte[testString.length() + 2];
readFully(is, buff);
assertEquals(testString, new String(buff, 0, testString.length()));
assertEquals('\r', buff[testString.length()]);
assertEquals('\n', buff[testString.length() + 1]);
server.close();
}
@Test
public void testWriteSerialized() throws Exception {
final int port = SocketUtils.findAvailableServerSocket();
final String testString = "abcdef";
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
server.setSoTimeout(10000);
Thread t = new Thread(new Runnable() {
public void run() {
try {
Socket socket = SocketFactory.getDefault().createSocket("localhost", port);
DefaultSerializer serializer = new DefaultSerializer();
serializer.serialize(testString, socket.getOutputStream());
serializer.serialize(testString, socket.getOutputStream());
Thread.sleep(1000000000L);
} catch (Exception e) {
e.printStackTrace();
}
}
});
t.setDaemon(true);
t.start();
Socket socket = server.accept();
socket.setSoTimeout(5000);
InputStream is = socket.getInputStream();
ObjectInputStream ois = new ObjectInputStream(is);
assertEquals(testString, ois.readObject());
ois = new ObjectInputStream(is);
assertEquals(testString, ois.readObject());
server.close();
}
/**
* @param is
* @param buff
*/
private void readFully(InputStream is, byte[] buff) throws IOException {
for (int i = 0; i < buff.length; i++) {
buff[i] = (byte) is.read();
}
}
}