From b339157a34f4f435e01dbfc9f3bd33d464abf0e0 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Tue, 28 Sep 2010 22:25:04 -0400 Subject: [PATCH] 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(); - } - } - -}