INT-1785 Add Raw Single-Shot (De)Serializer

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.<p/>
+ * 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;
+	}
+
+}

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()));
+	}
 }

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

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

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