From b339157a34f4f435e01dbfc9f3bd33d464abf0e0 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Tue, 28 Sep 2010 22:25:04 -0400 Subject: [PATCH 1/2] INT-1486 polish; rename byte array 'serializers' and all test usage --- .../AbstractClientConnectionFactory.java | 2 +- .../connection/AbstractConnectionFactory.java | 6 +- .../AbstractServerConnectionFactory.java | 2 +- .../tcp/connection/AbstractTcpConnection.java | 6 +- .../ip/tcp/connection/TcpConnection.java | 4 +- .../ip/tcp/connection/TcpMessageMapper.java | 2 +- .../ip/tcp/connection/TcpNetConnection.java | 5 +- .../ip/tcp/connection/TcpNioConnection.java | 2 +- .../AbstractByteArrayStreamingConverter.java | 67 ------ .../tcp/converter/ByteArrayCrLfConverter.java | 75 ------ .../ByteArrayLengthHeaderConverter.java | 113 --------- .../converter/ByteArrayStxEtxConverter.java | 80 ------- .../converter/SoftEndOfStreamException.java | 41 ---- .../ip/tcp/converter/package-info.java | 6 - .../ip/config/spring-integration-ip-2.0.xsd | 4 +- .../ip/tcp/TcpConfigInboundGatewayTests.java | 10 +- .../tcp/TcpReceivingChannelAdapterTests.java | 56 ++--- .../ip/tcp/TcpSendingMessageHandlerTests.java | 72 +++--- .../integration/ip/tcp/common-context.xml | 54 ++--- .../connection/TcpNioConnectionReadTests.java | 72 +++--- .../TcpNioConnectionWriteTests.java | 46 ++-- .../tcp/converter/DeserializationTests.java | 216 ------------------ .../ip/tcp/converter/SerializationTests.java | 182 --------------- 23 files changed, 172 insertions(+), 951 deletions(-) delete mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/AbstractByteArrayStreamingConverter.java delete mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/ByteArrayCrLfConverter.java delete mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/ByteArrayLengthHeaderConverter.java delete mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/ByteArrayStxEtxConverter.java delete mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/SoftEndOfStreamException.java delete mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/package-info.java delete mode 100644 spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/converter/DeserializationTests.java delete mode 100644 spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/converter/SerializationTests.java diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractClientConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractClientConnectionFactory.java index f8e476a3eb..0e2cf1d7b2 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractClientConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractClientConnectionFactory.java @@ -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 diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java index 8a45713703..d66727da1a 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java @@ -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(); diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractServerConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractServerConnectionFactory.java index d434b26b03..cbb9724e07 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractServerConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractServerConnectionFactory.java @@ -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. diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnection.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnection.java index 4b0d914267..f9b2580dcb 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnection.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnection.java @@ -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); } } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnection.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnection.java index 46de065050..f2ee28ade8 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnection.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnection.java @@ -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 diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapper.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapper.java index 13d2f229ef..56489a706a 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapper.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapper.java @@ -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; diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java index 2b4e78b16d..c33deea6c3 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java @@ -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, diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java index fd310d2a18..aa6b558132 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java @@ -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}. diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/AbstractByteArrayStreamingConverter.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/AbstractByteArrayStreamingConverter.java deleted file mode 100644 index 9a5037abd8..0000000000 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/AbstractByteArrayStreamingConverter.java +++ /dev/null @@ -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, - Deserializer { - - 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"); - } - } - -} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/ByteArrayCrLfConverter.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/ByteArrayCrLfConverter.java deleted file mode 100644 index 7e45571ad1..0000000000 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/ByteArrayCrLfConverter.java +++ /dev/null @@ -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(); - } - -} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/ByteArrayLengthHeaderConverter.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/ByteArrayLengthHeaderConverter.java deleted file mode 100644 index d95cdedd5d..0000000000 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/ByteArrayLengthHeaderConverter.java +++ /dev/null @@ -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; - } - -} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/ByteArrayStxEtxConverter.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/ByteArrayStxEtxConverter.java deleted file mode 100644 index 0029989fef..0000000000 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/ByteArrayStxEtxConverter.java +++ /dev/null @@ -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 <stx> and - * terminated by <etx> (not included in resulting byte[]). - * Writes a byte[] to an OutputStream and prefixed by <stx> terminated by <etx> - * - * @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(); - } - -} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/SoftEndOfStreamException.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/SoftEndOfStreamException.java deleted file mode 100644 index 9284268fe0..0000000000 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/SoftEndOfStreamException.java +++ /dev/null @@ -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); - } - -} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/package-info.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/package-info.java deleted file mode 100644 index ebcec3651d..0000000000 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/package-info.java +++ /dev/null @@ -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; - diff --git a/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.0.xsd b/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.0.xsd index 63b513d5f5..f1a8098a8b 100644 --- a/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.0.xsd +++ b/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.0.xsd @@ -299,7 +299,7 @@ the factory, the connection will be closed after a response is received. 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. @@ -313,7 +313,7 @@ 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 +associated with the connection. Default is ByteArrayCrLfSerializer. Serializer and Deserializer would normally be the same but this is not required. diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpConfigInboundGatewayTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpConfigInboundGatewayTests.java index 9f43bb75f3..4a39440e8d 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpConfigInboundGatewayTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpConfigInboundGatewayTests.java @@ -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); diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java index 21dc3faaa7..22d42e2b48 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java @@ -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(); diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java index a5ebd338a3..072d3cbc4a 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java @@ -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)); diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/common-context.xml b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/common-context.xml index 15eb1eeafb..a76a82ad12 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/common-context.xml +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/common-context.xml @@ -12,29 +12,29 @@ - - - + + + + serializer="crLfSerializer" + deserializer="crLfSerializer"/> + serializer="stxEtxSerializer" + deserializer="stxEtxSerializer"/> + serializer="lengthHeaderSerializer" + deserializer="lengthHeaderSerializer"/> + serializer="crLfSerializer" + deserializer="crLfSerializer"/> + serializer="stxEtxSerializer" + deserializer="stxEtxSerializer"/> + serializer="lengthHeaderSerializer" + deserializer="lengthHeaderSerializer"/> > responses = new ArrayList>(); 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> responses = new ArrayList>(); 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> responses = new ArrayList>(); 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> responses = new ArrayList>(); 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> responses = new ArrayList>(); final Semaphore semaphore = new Semaphore(0); final List added = new ArrayList(); final List removed = new ArrayList(); - 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> responses = new ArrayList>(); final Semaphore semaphore = new Semaphore(0); final List added = new ArrayList(); final List removed = new ArrayList(); - 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> responses = new ArrayList>(); final Semaphore semaphore = new Semaphore(0); final List added = new ArrayList(); final List removed = new ArrayList(); - 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> responses = new ArrayList>(); final Semaphore semaphore = new Semaphore(0); final List added = new ArrayList(); final List removed = new ArrayList(); - 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 added = new ArrayList(); final List removed = new ArrayList(); - AbstractServerConnectionFactory scf = getConnectionFactory(port, converter,new TcpListener() { + AbstractServerConnectionFactory scf = getConnectionFactory(port, serializer,new TcpListener() { public boolean onMessage(Message message) { responses.add(message); return false; diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionWriteTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionWriteTests.java index 9bf22f7d46..f4e07efed2 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionWriteTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionWriteTests.java @@ -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); diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/converter/DeserializationTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/converter/DeserializationTests.java deleted file mode 100644 index fa3abce0fd..0000000000 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/converter/DeserializationTests.java +++ /dev/null @@ -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(); - } - -} diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/converter/SerializationTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/converter/SerializationTests.java deleted file mode 100644 index e6ada7fe9b..0000000000 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/converter/SerializationTests.java +++ /dev/null @@ -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(); - } - } - -} From 3c93433bafb26d31e2366856d6e788eefeeff948 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Tue, 28 Sep 2010 22:25:54 -0400 Subject: [PATCH 2/2] INT-1486 polish; rename byte array 'serializers' and all test usage --- .../AbstractByteArraySerializer.java | 68 ++++++ .../serializer/ByteArrayCrLfSerializer.java | 75 ++++++ .../ByteArrayLengthHeaderSerializer.java | 113 +++++++++ .../serializer/ByteArrayStxEtxSerializer.java | 80 +++++++ .../serializer/SoftEndOfStreamException.java | 41 ++++ .../ip/tcp/serializer/package-info.java | 6 + .../tcp/serializer/DeserializationTests.java | 219 ++++++++++++++++++ .../ip/tcp/serializer/SerializationTests.java | 185 +++++++++++++++ 8 files changed, 787 insertions(+) create mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/AbstractByteArraySerializer.java create mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayCrLfSerializer.java create mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayLengthHeaderSerializer.java create mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayStxEtxSerializer.java create mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/SoftEndOfStreamException.java create mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/package-info.java create mode 100644 spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/serializer/DeserializationTests.java create mode 100644 spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/serializer/SerializationTests.java diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/AbstractByteArraySerializer.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/AbstractByteArraySerializer.java new file mode 100644 index 0000000000..b94e3e00f5 --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/AbstractByteArraySerializer.java @@ -0,0 +1,68 @@ +/* + * 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.serializer; + +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 (de)serializers that provide a mechanism to + * reconstruct a byte array from an arbitrary stream. + * + * @author Gary Russell + * @since 2.0 + * + */ +public abstract class AbstractByteArraySerializer implements + Serializer, + Deserializer { + + protected int maxMessageSize = 2048; + + protected Log logger = LogFactory.getLog(this.getClass()); + + /** + * The maximum supported message size for this serializer. + * Default 2048. + * @return The max message size. + */ + public int getMaxMessageSize() { + return maxMessageSize; + } + + /** + * The maximum supported message size for this serializer. + * 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"); + } + } + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayCrLfSerializer.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayCrLfSerializer.java new file mode 100644 index 0000000000..72fda36108 --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayCrLfSerializer.java @@ -0,0 +1,75 @@ +/* + * 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.serializer; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; + +/** + * Reads data in an InputStream to a byte[]; data must be terminated by \r\n + * (not included in resulting byte[]). + * Writes a byte[] to an OutputStream and adds \r\n. + * + * @author Gary Russell + * @since 2.0 + */ +public class ByteArrayCrLfSerializer extends AbstractByteArraySerializer { + + /** + * Reads the data in the inputstream to a byte[]. Data must be terminated + * by CRLF (\r\n). Throws a {@link SoftEndOfStreamException} if the stream + * is closed immediately after the \r\n (i.e. no data is in the process of + * being read). + */ + 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(); + } + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayLengthHeaderSerializer.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayLengthHeaderSerializer.java new file mode 100644 index 0000000000..d76067d8f2 --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayLengthHeaderSerializer.java @@ -0,0 +1,113 @@ +/* + * 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.serializer; + +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; + +/** + * Reads data in an InputStream to a byte[]; data must be preceded by + * a 4 byte binary length (network byte order, + * not included in resulting byte[]). + * Writes a byte[] to an OutputStream after a 4 byte binary length. + * The length field contains the length of data following the length + * field. + * (network byte order). + * + * @author Gary Russell + * @since 2.0 + */ +public class ByteArrayLengthHeaderSerializer extends AbstractByteArraySerializer { + + 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; + } + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayStxEtxSerializer.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayStxEtxSerializer.java new file mode 100644 index 0000000000..1a6ab9c8bc --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayStxEtxSerializer.java @@ -0,0 +1,80 @@ +/* + * 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.serializer; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; + +import org.springframework.integration.mapping.MessageMappingException; + +/** + * Reads data in an InputStream to a byte[]; data must be prefixed by <stx> and + * terminated by <etx> (not included in resulting byte[]). + * Writes a byte[] to an OutputStream prefixed by <stx> terminated by <etx> + * + * @author Gary Russell + * @since 2.0 + */ +public class ByteArrayStxEtxSerializer extends AbstractByteArraySerializer { + + public static final int STX = 0x02; + + public static final int ETX = 0x03; + + /** + * Reads the data in the inputstream to a byte[]. Data must be prefixed + * with an ASCII STX character, and terminated with an ASCII ETX character. + * Throws a {@link SoftEndOfStreamException} if the stream + * is closed immediately before the STX (i.e. no data is in the process of + * 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(); + } + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/SoftEndOfStreamException.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/SoftEndOfStreamException.java new file mode 100644 index 0000000000..c91b445247 --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/SoftEndOfStreamException.java @@ -0,0 +1,41 @@ +/* + * 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.serializer; + +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); + } + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/package-info.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/package-info.java new file mode 100644 index 0000000000..7446699fda --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/package-info.java @@ -0,0 +1,6 @@ +/** + * Byte array (de)serializers for putting some protocol on the + * wire so that incoming messages can be constructed from stream data. + */ +package org.springframework.integration.ip.tcp.serializer; + diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/serializer/DeserializationTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/serializer/DeserializationTests.java new file mode 100644 index 0000000000..2f3a73c882 --- /dev/null +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/serializer/DeserializationTests.java @@ -0,0 +1,219 @@ +/* + * 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.serializer; + +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.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; + +/** + * @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); + ByteArrayLengthHeaderSerializer serializer = new ByteArrayLengthHeaderSerializer(); + byte[] out = serializer.deserialize(socket.getInputStream()); + assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, + new String(out)); + out = serializer.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); + ByteArrayStxEtxSerializer serializer = new ByteArrayStxEtxSerializer(); + byte[] out = serializer.deserialize(socket.getInputStream()); + assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, + new String(out)); + out = serializer.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); + ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer(); + byte[] out = serializer.deserialize(socket.getInputStream()); + assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, + new String(out)); + out = serializer.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); + ByteArrayLengthHeaderSerializer serializer = new ByteArrayLengthHeaderSerializer(); + try { + serializer.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); + ByteArrayStxEtxSerializer serializer = new ByteArrayStxEtxSerializer(); + try { + serializer.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); + ByteArrayStxEtxSerializer serializer = new ByteArrayStxEtxSerializer(); + serializer.setMaxMessageSize(1024); + try { + serializer.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); + ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer(); + try { + serializer.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); + ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer(); + serializer.setMaxMessageSize(1024); + try { + serializer.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(); + } + +} diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/serializer/SerializationTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/serializer/SerializationTests.java new file mode 100644 index 0000000000..084fabfa0a --- /dev/null +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/serializer/SerializationTests.java @@ -0,0 +1,185 @@ +/* + * 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.serializer; + +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.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; + +/** + * @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()); + ByteArrayLengthHeaderSerializer serializer = new ByteArrayLengthHeaderSerializer(); + serializer.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()); + ByteArrayStxEtxSerializer serializer = new ByteArrayStxEtxSerializer(); + serializer.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(ByteArrayStxEtxSerializer.STX, buff[0]); + assertEquals(testString, new String(buff, 1, testString.length())); + assertEquals(ByteArrayStxEtxSerializer.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()); + ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer(); + serializer.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(); + } + } + +}