diff --git a/docs/src/reference/docbook/ip.xml b/docs/src/reference/docbook/ip.xml index 3af13faa1e..3bf14e07bf 100644 --- a/docs/src/reference/docbook/ip.xml +++ b/docs/src/reference/docbook/ip.xml @@ -217,14 +217,42 @@ Connection factories are configured to use (de)serializers to convert between the message payload and the bits that are sent over TCP. This is accomplished by providing a deserializer and serializer for inbound and outbound messages respectively. - Four standard (de)serializers are provided; the first is ByteArrayCrlfSerializer, - which can convert a byte array to a stream of bytes followed by carriage + A number of standard (de)serializers are provided. + + + The ByteArrayCrlfSerializer, + converts a byte array to a stream of bytes followed by carriage return and linefeed characters (\r\n). This is the default (de)serializer and can be used with - telnet as a client, for example. The second is is ByteArrayStxEtxSerializer, - which can convert a byte array to a stream of bytes preceded by an STX (0x02) and - followed by an ETX (0x03). The third is ByteArrayLengthHeaderSerializer, - which can convert a byte array to a stream of bytes preceded by a 4 byte binary - length in network byte order. Each of these is a subclass of + telnet as a client, for example. + + + The ByteArrayStxEtxSerializer, + converts a byte array to a stream of bytes preceded by an STX (0x02) and + followed by an ETX (0x03). + + + The ByteArrayLengthHeaderSerializer, + converts a byte array to a stream of bytes preceded by a 4 byte binary + length in network byte order. This a very efficient deserializer + because it does not have to parse every byte looking for a termination + character sequence. It can also be used for payloads containing binary data; + the above serializers only support text in the payload. + + + The ByteArrayRawSerializer, + converts a byte array to a stream of bytes and adds no additional message + demarcation data; with this (de)serializer, the end of a message is indicated + by the client closing the socket in an orderly fashion. When using this serializer, + message reception will hang until the client closes the socket, or a timeout occurs; + a timeout will NOT result in a message. When this serializer is being used, and the client + is a Spring Integration application, the client must use a connection factory that is + configured with single-use=true - this causes the adapter to close the socket after sending + the message; the serializer will not, itself, close the connection. This serializer + should only be used with connection factories used by channel adapters (not gateways), and the + connection factories should be used by either an inbound or outbound adapter, and not both. + + + Each of these is a subclass of AbstractByteArraySerializer which implements both org.springframework.core.serializer.Serializer and org.springframework.core.serializer.Deserializer. @@ -232,11 +260,25 @@ AbstractByteArraySerializer for serialization will also accept a String which will be converted to a byte array first. Each of these (de)serializers converts an input stream containing the - corresponding format to a byte array payload. The fourth standard serializer is + corresponding format to a byte array payload. + + + To avoid memory exhaustion due to a badly behaved client (one that does not adhere to + the protocol of the configured serializer), these serializers impose a maximum message + size. If the size is exceeded by an incoming message, an exception will be thrown. + The default maximum message size is 2048 bytes, and can be increased by setting the + maxMessageSize property. If you are using the default (de)serializer + and wish to increase the maximum message size, you must declare it as an explicit bean + with the property set and configure the connection factory to use that bean. + + + The final standard serializer is org.springframework.core.serializer.DefaultSerializer which can be used to convert Serializable objects using java serialization. org.springframework.core.serializer.DefaultDeserializer is provided for inbound deserialization of streams containing Serializable objects. + + To implement a custom (de)serializer pair, implement the org.springframework.core.serializer.Deserializer and org.springframework.core.serializer.Serializer interfaces. If you do not wish to use @@ -246,16 +288,16 @@ - + + ]]> A server connection factory that uses java.net.Socket connections and uses Java serialization on the wire. @@ -632,7 +674,7 @@ Y Y true, false - Whether or not the socket handing uses NIO. Refer to the java.nio + Whether or not connection uses NIO. Refer to the java.nio package for more information. See . Default false. @@ -642,7 +684,7 @@ Y N true, false - When using NIO, whether or not the tcp adapter uses direct buffers. + When using NIO, whether or not the connection uses direct buffers. Refer to java.nio.ByteBuffer documentation for more information. Must be false if using-nio is false. @@ -745,7 +787,7 @@ Y Y - Documentation to be supplied. + See diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayRawSerializer.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayRawSerializer.java new file mode 100644 index 0000000000..cb6c1698d7 --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayRawSerializer.java @@ -0,0 +1,68 @@ +/* + * Copyright 2002-2011 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; + +/** + * A byte array (de)serializer that does nothing with the payload; sends it raw. + * Message termination for assembly purposes is signaled by the client closing the + * connection. The serializer does not, itself, close the connection after + * writing the bytes.

+ * Because the socket must be closed to indicate message end, this (de)serializer + * can only be used by uni-directional (non-collaborating) channel adapters, and + * not by gateways. + * + * @author Gary Russell + * @since 2.0.3 + * + */ +public class ByteArrayRawSerializer extends AbstractByteArraySerializer { + + public void serialize(byte[] bytes, OutputStream outputStream) + throws IOException { + outputStream.write(bytes); + outputStream.flush(); + } + + public byte[] deserialize(InputStream inputStream) throws IOException { + byte[] buffer = new byte[this.maxMessageSize]; + int n = 0; + int bite = 0; + if (logger.isDebugEnabled()) + logger.debug("Available to read:" + inputStream.available()); + while (bite >= 0) { + bite = inputStream.read(); + if (bite < 0) { + if (n == 0) { + throw new SoftEndOfStreamException("Stream closed between payloads"); + } + break; + } + buffer[n++] = (byte) bite; + if (n >= this.maxMessageSize) { + throw new IOException("Socket was not closed before max message length: " + + this.maxMessageSize); + } + }; + byte[] assembledData = new byte[n]; + System.arraycopy(buffer, 0, assembledData, 0, n); + return assembledData; + } + +} diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests.java index 1d791264d0..e12e788e8c 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests.java @@ -32,6 +32,7 @@ import org.springframework.integration.history.MessageHistory; import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory; import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory; import org.springframework.integration.ip.tcp.connection.TcpConnection; +import org.springframework.integration.ip.tcp.serializer.ByteArrayRawSerializer; import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.test.util.TestUtils; import org.springframework.test.context.ContextConfiguration; @@ -80,4 +81,20 @@ public class ConnectionToConnectionTests { assertEquals("Test", new String((byte[]) message.getPayload())); } + @Test + public void testConnectRaw() throws Exception { + ByteArrayRawSerializer serializer = new ByteArrayRawSerializer(); + client.setSerializer(serializer); + server.setDeserializer(serializer); + TcpConnection connection = client.getConnection(); + connection.send(MessageBuilder.withPayload("Test").build()); + Message message = serverSideChannel.receive(10000); + MessageHistory history = MessageHistory.read(message); + //org.springframework.integration.test.util.TestUtils + Properties componentHistoryRecord = TestUtils.locateComponentInHistory(history, "looper", 0); + assertNotNull(componentHistoryRecord); + assertTrue(componentHistoryRecord.get("type").equals("ip:tcp-inbound-gateway")); + assertNotNull(message); + assertEquals("Test", new String((byte[]) message.getPayload())); + } } 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 index 34d3fb12d6..4d5dab20b1 100644 --- 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 @@ -90,6 +90,21 @@ public class DeserializationTests { server.close(); } + @Test + public void testReadRaw() throws Exception { + int port = SocketTestUtils.findAvailableServerSocket(); + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + server.setSoTimeout(10000); + SocketTestUtils.testSendRaw(port); + Socket socket = server.accept(); + socket.setSoTimeout(5000); + ByteArrayRawSerializer serializer = new ByteArrayRawSerializer(); + byte[] out = serializer.deserialize(socket.getInputStream()); + assertEquals("Data", SocketTestUtils.TEST_STRING + SocketTestUtils.TEST_STRING, + new String(out)); + server.close(); + } + @Test public void testReadSerialized() throws Exception { int port = SocketTestUtils.findAvailableServerSocket(); 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 index d2cae47c07..1ff12564c0 100644 --- 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 @@ -138,6 +138,39 @@ public class SerializationTests { server.close(); } + @Test + public void testWriteRaw() throws Exception { + final int port = SocketTestUtils.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()); + ByteArrayRawSerializer serializer = new ByteArrayRawSerializer(); + serializer.serialize(buffer.array(), socket.getOutputStream()); + socket.close(); + 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() + 1]; + readFully(is, buff); + assertEquals(testString, new String(buff, 0, testString.length())); + assertEquals(-1, buff[testString.length()]); + server.close(); + } + @Test public void testWriteSerialized() throws Exception { final int port = SocketTestUtils.findAvailableServerSocket(); diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/util/SocketTestUtils.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/util/SocketTestUtils.java index 932356df8d..cbc8ea5909 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/util/SocketTestUtils.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/util/SocketTestUtils.java @@ -252,6 +252,26 @@ public class SocketTestUtils { thread.start(); } + /** + * Sends a single message in two chunks and then closes the socket. + */ + public static void testSendRaw(final int port) { + Thread thread = new Thread(new Runnable() { + public void run() { + try { + Socket socket = new Socket(InetAddress.getByName("localhost"), port); + OutputStream outputStream = socket.getOutputStream(); + outputStream.write(TEST_STRING.getBytes()); + outputStream.write(TEST_STRING.getBytes()); + socket.close(); + } catch (Exception e) { + e.printStackTrace(); + } + } + }); + thread.setDaemon(true); + thread.start(); + } /** * Sends two serialized objects over the same socket. * @param port