INT-1807 Add Mechanism For Headers with TCP

TCP streams have no standard message structure. Therefore, the
TCP implementation previously only transferred the message
payload.

If someone wanted to convey header information, they would have
to write their own wrapper and/or use Java serialization for
the entire message.

This change provides a strategy to allow users to determine
which headers are transferred, and how.

A MessageConvertingMessageMapper is now provided that invokes
any MessageConverter. A MapMessageConverter is provided that
converts the payload, and selected heades to a Map with two
entries ("payload") and ("headers").

A MapJsonSerializer is provided that converts a Map to/from
JSON. Jackson can't delimit multiple objects in a stream
so another serializer is required to encode/decode structure.
A ByteArrayLfSerializer is used by default, inserting a
linefeed between JSON objects.

The combination of these elements now allows header
information to be transferred over TCP. Of course, users
can implment their own (de)serializer to format the
bits on the wire exactly as needed by their application.

INT-1807 Polishing

Add a test that uses a Map MessageConverter with a
Java (de)serializer.

INT-1807: Polishing

INT-1807: Rebased and polished

Change `MapJsonSerializer` to use `JsonObjectMapper` abstraction

Doc Polishing

spring-integration-core/src/main/java/org/springframework/integration/support/converter/MapMessageConverter.java

@@ -0,0 +1,102 @@
+/*
+ * Copyright 2002-2013 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.support.converter;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Set;
+
+import org.springframework.integration.Message;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.util.Assert;
+
+/**
+ * Converts to/from a Map with 2 keys ('headers' and 'payload').
+ * @author Gary Russell
+ * @since 3.0
+ *
+ */
+public class MapMessageConverter implements MessageConverter {
+
+	private volatile String[] headerNames;
+
+	private volatile boolean filterHeadersInToMessage;
+
+	/**
+	 * Headers to be converted in {@link #fromMessage(Message)}.
+	 * {@link #toMessage(Object)} will populate all headers found in
+	 * the map, unless {@link #filterHeadersInToMessage} is true.
+	 * @param headerNames
+	 */
+	public void setHeaderNames(String... headerNames) {
+		this.headerNames = headerNames;
+	}
+
+	/**
+	 * By default all headers on Map passed to {@link #toMessage(Object)}
+	 * will be mapped. Set this property
+	 * to 'true' if you wish to limit the inbound headers to those in
+	 * the #headerNames.
+	 * @param filterHeadersInToMessage
+	 */
+	public void setFilterHeadersInToMessage(boolean filterHeadersInToMessage) {
+		this.filterHeadersInToMessage = filterHeadersInToMessage;
+	}
+
+	public <P> Message<P> toMessage(Object object) {
+		Assert.isInstanceOf(Map.class, object, "This converter expects a Map");
+		@SuppressWarnings("unchecked")
+		Map<String, ?> map = (Map<String, ?>) object;
+		Object payload = map.get("payload");
+		Assert.notNull(payload, "'payload' entry cannot be null");
+		MessageBuilder<?> messageBuilder = MessageBuilder.withPayload(payload);
+		@SuppressWarnings("unchecked")
+		Map<String, ?> headers = (Map<String, ?>) map.get("headers");
+		if (headers != null) {
+			if (this.filterHeadersInToMessage) {
+				headers.keySet().retainAll(Arrays.asList(this.headerNames));
+			}
+			messageBuilder.copyHeaders(headers);
+			/*for (Entry<String, ?> entry : headers.entrySet()) {
+				if (this.filterHeadersInToMessage ? this.headerNames.contains(entry.getKey()) : true) {
+					messageBuilder.setHeader(entry.getKey(), entry.getValue());
+				}
+			}*/
+		}
+		@SuppressWarnings("unchecked")
+		Message<P> convertedMessage = (Message<P>) messageBuilder.build();
+		return convertedMessage;
+	}
+
+	public <P> Object fromMessage(Message<P> message) {
+		Map<String,Object> map = new HashMap<String, Object>();
+		map.put("payload", message.getPayload());
+		Map<String, Object> headers = new HashMap<String, Object>();
+		for (String headerName : headerNames) {
+			Object header = message.getHeaders().get(headerName);
+			if (header != null) {
+				headers.put(headerName, header);
+			}
+		}
+		map.put("headers", headers);
+		return map;
+	}
+
+}

spring-integration-core/src/test/java/org/springframework/integration/support/converter/MapMessageConverterTests.java

@@ -0,0 +1,130 @@
+/*
+ * Copyright 2002-2013 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.support.converter;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.fail;
+
+import java.util.Map;
+
+import org.junit.Test;
+import org.springframework.integration.Message;
+import org.springframework.integration.support.MessageBuilder;
+
+/**
+ * @author Gary Russell
+ * @author Artem Bilan
+ * @since 3.0
+ *
+ */
+public class MapMessageConverterTests {
+
+	@Test
+	public void testFromMessageToMessage() throws Exception {
+		Message<String> message = MessageBuilder.withPayload("foo")
+				.setHeader("bar", "baz")
+				.setHeader("baz", "qux")
+				.build();
+		MapMessageConverter converter = new MapMessageConverter();
+		converter.setHeaderNames("bar");
+		@SuppressWarnings("unchecked")
+		Map<String, Object> map =  (Map<String, Object>) converter.fromMessage(message);
+		@SuppressWarnings("unchecked")
+		Map<String, Object> headers = (Map<String, Object>) map.get("headers");
+
+		assertNotNull(headers);
+		assertNotNull(map.get("payload"));
+		assertEquals("foo", map.get("payload"));
+		assertNotNull(headers.get("bar"));
+		assertEquals("baz", headers.get("bar"));
+		assertNull(headers.get("baz"));
+
+		headers.put("baz", "qux");
+		message = converter.toMessage(map);
+		assertEquals("foo", message.getPayload());
+		assertEquals("baz", message.getHeaders().get("bar"));
+		assertEquals("qux", message.getHeaders().get("baz"));
+
+		converter.setFilterHeadersInToMessage(true);
+
+		message = converter.toMessage(map);
+		assertEquals("foo", message.getPayload());
+		assertEquals("baz", message.getHeaders().get("bar"));
+		assertNull(message.getHeaders().get("baz"));
+	}
+
+	@Test
+	public void testInvalid() throws Exception {
+		Message<String> message = MessageBuilder.withPayload("foo")
+				.setHeader("bar", "baz")
+				.setHeader("baz", "qux")
+				.build();
+		MapMessageConverter converter = new MapMessageConverter();
+		converter.setHeaderNames("bar");
+		@SuppressWarnings("unchecked")
+		Map<String, Object> map =  (Map<String, Object>) converter.fromMessage(message);
+
+		map.remove("payload");
+
+		try {
+			converter.toMessage(map);
+			fail("Expected exception");
+		}
+		catch (IllegalArgumentException e) {
+			assertEquals("'payload' entry cannot be null", e.getMessage());
+		}
+	}
+
+	@Test
+	public void testNoHeaders() throws Exception {
+		Message<String> message = MessageBuilder.withPayload("foo")
+				.build();
+		MapMessageConverter converter = new MapMessageConverter();
+		converter.setHeaderNames("bar");
+		@SuppressWarnings("unchecked")
+		Map<String, Object> map =  (Map<String, Object>) converter.fromMessage(message);
+		@SuppressWarnings("unchecked")
+		Map<String, Object> headers = (Map<String, Object>) map.get("headers");
+
+		assertNotNull(headers);
+		assertEquals(0, headers.size());
+		map.remove("headers");
+		message = converter.toMessage(map);
+		assertEquals("foo", message.getPayload());
+	}
+
+	@Test
+	public void testNotIncludedIfNull() throws Exception {
+		Message<String> message = MessageBuilder.withPayload("foo")
+				.setHeader("bar", null)
+				.build();
+		MapMessageConverter converter = new MapMessageConverter();
+		converter.setHeaderNames("bar");
+		@SuppressWarnings("unchecked")
+		Map<String, Object> map =  (Map<String, Object>) converter.fromMessage(message);
+		@SuppressWarnings("unchecked")
+		Map<String, Object> headers = (Map<String, Object>) map.get("headers");
+
+		assertNotNull(headers);
+		assertNotNull(map.get("payload"));
+		assertEquals("foo", map.get("payload"));
+		assertFalse(headers.keySet().contains("bar"));
+		assertEquals(0, headers.size());
+	}
+}

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/MessageConvertingTcpMessageMapper.java

@@ -0,0 +1,60 @@
+/*
+ * Copyright 2002-2013 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.connection;
+
+import org.springframework.integration.Message;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.integration.support.converter.MessageConverter;
+import org.springframework.util.Assert;
+
+/**
+ * @author Gary Russell
+ * @since 3.0
+ *
+ */
+public class MessageConvertingTcpMessageMapper extends TcpMessageMapper {
+
+	private final MessageConverter messageConverter;
+
+	public MessageConvertingTcpMessageMapper(MessageConverter messageConverter) {
+		Assert.notNull(messageConverter, "'messasgeConverter' must not be null");
+		this.messageConverter = messageConverter;
+	}
+
+	@Override
+	public Message<Object> toMessage(TcpConnection connection) throws Exception {
+		Object data = connection.getPayload();
+		if (data != null) {
+			Message<Object> message = this.messageConverter.toMessage(data);
+			MessageBuilder<Object> messageBuilder = MessageBuilder.fromMessage(message);
+			this.addStandardHeaders(connection, messageBuilder);
+			this.addCustomHeaders(connection, messageBuilder);
+			return messageBuilder.build();
+		}
+		else {
+			if (logger.isWarnEnabled()) {
+				logger.warn("Null payload from connection " + connection.getConnectionId());
+			}
+			return null;
+		}
+	}
+
+	@Override
+	public Object fromMessage(Message<?> message) throws Exception {
+		return this.messageConverter.fromMessage(message);
+	}
+
+}

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapper.java

@@ -18,6 +18,8 @@ package org.springframework.integration.ip.tcp.connection;
 import java.io.UnsupportedEncodingException;
 import java.util.Map;
 
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
 import org.springframework.integration.Message;
 import org.springframework.integration.MessageHandlingException;
 import org.springframework.integration.ip.IpHeaders;
@@ -43,6 +45,8 @@ public class TcpMessageMapper implements
 		InboundMessageMapper<TcpConnection>,
 		OutboundMessageMapper<Object> {
 
+	protected final Log logger = LogFactory.getLog(this.getClass());
+
 	private volatile String charset = "UTF-8";
 
 	private volatile boolean stringToBytes = true;
@@ -54,26 +58,39 @@ public class TcpMessageMapper implements
 		Object payload = connection.getPayload();
 		if (payload != null) {
 			MessageBuilder<Object> messageBuilder = MessageBuilder.withPayload(payload);
-			String connectionId = connection.getConnectionId();
-			messageBuilder
-				.setHeader(IpHeaders.HOSTNAME, connection.getHostName())
-				.setHeader(IpHeaders.IP_ADDRESS, connection.getHostAddress())
-				.setHeader(IpHeaders.REMOTE_PORT, connection.getPort())
-				.setHeader(IpHeaders.CONNECTION_ID, connectionId);
-			if (this.applySequence) {
-				messageBuilder
-					.setCorrelationId(connectionId)
-					.setSequenceNumber((int) connection.incrementAndGetConnectionSequence());
-			}
-			Map<String, ?> customHeaders = this.supplyCustomHeaders(connection);
-			if (customHeaders != null) {
-				messageBuilder.copyHeadersIfAbsent(customHeaders);
-			}
+			this.addStandardHeaders(connection, messageBuilder);
+			this.addCustomHeaders(connection, messageBuilder);
 			message = messageBuilder.build();
 		}
+		else {
+			if (logger.isWarnEnabled()) {
+				logger.warn("Null payload from connection " + connection.getConnectionId());
+			}
+		}
 		return message;
 	}
 
+	protected final void addStandardHeaders(TcpConnection connection, MessageBuilder<?> messageBuilder) {
+		String connectionId = connection.getConnectionId();
+		messageBuilder
+			.setHeader(IpHeaders.HOSTNAME, connection.getHostName())
+			.setHeader(IpHeaders.IP_ADDRESS, connection.getHostAddress())
+			.setHeader(IpHeaders.REMOTE_PORT, connection.getPort())
+			.setHeader(IpHeaders.CONNECTION_ID, connectionId);
+		if (this.applySequence) {
+			messageBuilder
+				.setCorrelationId(connectionId)
+				.setSequenceNumber((int) connection.incrementAndGetConnectionSequence());
+		}
+	}
+
+	protected final void addCustomHeaders(TcpConnection connection, MessageBuilder<?> messageBuilder) {
+		Map<String, ?> customHeaders = this.supplyCustomHeaders(connection);
+		if (customHeaders != null) {
+			messageBuilder.copyHeadersIfAbsent(customHeaders);
+		}
+	}
+
 	/**
 	 * Override to provide additional headers. The standard headers cannot be overridden
 	 * and any such headers will be ignored if provided in the result.

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/MapJsonSerializer.java

@@ -0,0 +1,111 @@
+/*
+ * Copyright 2002-2013 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.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.io.OutputStream;
+import java.io.OutputStreamWriter;
+import java.util.Map;
+
+import org.springframework.core.serializer.Deserializer;
+import org.springframework.core.serializer.Serializer;
+import org.springframework.integration.support.json.JacksonJsonObjectMapperProvider;
+import org.springframework.integration.support.json.JsonObjectMapper;
+import org.springframework.util.Assert;
+
+/**
+ * Serializes a {@link Map} as JSON. Deserializes JSON to
+ * a {@link Map}. The default {@link JacksonJsonObjectMapperProvider#newInstance()} can be
+ * overridden using {@link #setJsonObjectMapper(JsonObjectMapper)}.
+ * <p/>
+ * The JSON deserializer can't delimit multiple JSON
+ * objects. Therefore another (de)serializer is used to
+ * apply structure to the stream. By default, this is a
+ * simple {@link ByteArrayLfSerializer}, which inserts/expects
+ * LF (0x0a) between messages.
+ *
+ * @author Gary Russell
+ * @author Artem Bilan
+ * @since 3.0
+ *
+ */
+public class MapJsonSerializer implements Serializer<Map<?, ?>>, Deserializer<Map<?, ?>> {
+
+	private volatile JsonObjectMapper<?> jsonObjectMapper = JacksonJsonObjectMapperProvider.newInstance();
+
+	private volatile Deserializer<byte[]> packetDeserializer = new ByteArrayLfSerializer();
+
+	private volatile Serializer<byte[]> packetSerializer = new ByteArrayLfSerializer();
+
+	/**
+	 * An {@link JsonObjectMapper} to be used for the conversion to/from
+	 * JSON. Use this if you wish to set additional {@link JsonObjectMapper} implementation features.
+	 * @param jsonObjectMapper the jsonObjectMapper.
+	 */
+	public void setJsonObjectMapper(JsonObjectMapper<?> jsonObjectMapper) {
+		Assert.notNull(jsonObjectMapper, "'jsonObjectMapper' cannot be null");
+		this.jsonObjectMapper = jsonObjectMapper;
+	}
+
+	/**
+	 * A {@link Deserializer} that will construct the full JSON content from
+	 * the stream which is then passed to the JsonObjectMapper. Default is
+	 * {@link ByteArrayLfSerializer}.
+	 * @param packetDeserializer the packetDeserializer
+	 */
+	public void setPacketDeserializer(Deserializer<byte[]> packetDeserializer) {
+		Assert.notNull(packetDeserializer, "'packetDeserializer' cannot be null");
+		this.packetDeserializer = packetDeserializer;
+	}
+
+	/**
+	 * A {@link Serializer} that will delimit the full JSON content in
+	 * the stream. Default is
+	 * {@link ByteArrayLfSerializer}.
+	 * @param packetSerializer the packetSerializer
+	 */
+	public void setPacketSerializer(Serializer<byte[]> packetSerializer) {
+		Assert.notNull(packetSerializer, "'packetSerializer' cannot be null");
+		this.packetSerializer = packetSerializer;
+	}
+
+	public Map<?, ?> deserialize(InputStream inputStream) throws IOException {
+		byte[] bytes = this.packetDeserializer.deserialize(inputStream);
+		try {
+			return this.jsonObjectMapper.fromJson(new InputStreamReader(new ByteArrayInputStream(bytes)), Map.class);
+		}
+		catch (Exception e) {
+			throw new IOException(e);
+		}
+	}
+
+	public void serialize(Map<?, ?> object, OutputStream outputStream) throws IOException {
+		ByteArrayOutputStream baos = new ByteArrayOutputStream();
+		try {
+			this.jsonObjectMapper.toJson(object, new OutputStreamWriter(baos));
+		}
+		catch (Exception e) {
+			throw new IOException(e);
+		}
+		this.packetSerializer.serialize(baos.toByteArray(), outputStream);
+		outputStream.flush();
+	}
+
+}

spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapperTests.java

@@ -20,6 +20,8 @@ import static org.junit.Assert.assertNotNull;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
 import java.net.Socket;
 import java.util.Collections;
 import java.util.Map;
@@ -27,9 +29,13 @@ import java.util.Map;
 import javax.net.SocketFactory;
 
 import org.junit.Test;
+import org.springframework.core.serializer.DefaultDeserializer;
+import org.springframework.core.serializer.DefaultSerializer;
 import org.springframework.integration.Message;
 import org.springframework.integration.ip.IpHeaders;
+import org.springframework.integration.ip.tcp.serializer.MapJsonSerializer;
 import org.springframework.integration.support.MessageBuilder;
+import org.springframework.integration.support.converter.MapMessageConverter;
 
 /**
  * @author Gary Russell
@@ -207,4 +213,72 @@ public class TcpMessageMapperTests {
 
 	}
 
-}
+	@Test
+	public void testMapMessageConvertingOutboundJson() throws Exception {
+		Message<String> message = MessageBuilder.withPayload("foo")
+				.setHeader("bar", "baz")
+				.build();
+		MapMessageConverter converter = new MapMessageConverter();
+		converter.setHeaderNames("bar");
+		MessageConvertingTcpMessageMapper mapper = new MessageConvertingTcpMessageMapper(converter);
+		Map<?, ?> map = (Map<?, ?>) mapper.fromMessage(message);
+		MapJsonSerializer serializer = new MapJsonSerializer();
+		ByteArrayOutputStream baos = new ByteArrayOutputStream();
+		serializer.serialize(map, baos);
+		assertEquals("{\"headers\":{\"bar\":\"baz\"},\"payload\":\"foo\"}\n", new String(baos.toByteArray(), "UTF-8"));
+	}
+
+	@Test
+	public void testMapMessageConvertingInboundJson() throws Exception {
+		String json = "{\"headers\":{\"bar\":\"baz\"},\"payload\":\"foo\"}\n";
+		MapMessageConverter converter = new MapMessageConverter();
+		MessageConvertingTcpMessageMapper mapper = new MessageConvertingTcpMessageMapper(converter);
+		MapJsonSerializer deserializer = new MapJsonSerializer();
+		Map<?, ?> map = deserializer.deserialize(new ByteArrayInputStream(json.getBytes("UTF-8")));
+
+		TcpConnection connection = mock(TcpConnection.class);
+		when(connection.getPayload()).thenReturn(map);
+		when(connection.getHostName()).thenReturn("someHost");
+		when(connection.getHostAddress()).thenReturn("1.1.1.1");
+		when(connection.getPort()).thenReturn(1234);
+		when(connection.getConnectionId()).thenReturn("someId");
+		Message<?> message = mapper.toMessage(connection);
+		assertEquals("foo", message.getPayload());
+		assertEquals("baz", message.getHeaders().get("bar"));
+		assertEquals("someHost", message.getHeaders().get(IpHeaders.HOSTNAME));
+		assertEquals("1.1.1.1", message.getHeaders().get(IpHeaders.IP_ADDRESS));
+		assertEquals(1234, message.getHeaders().get(IpHeaders.REMOTE_PORT));
+		assertEquals("someId", message.getHeaders().get(IpHeaders.CONNECTION_ID));
+	}
+
+	@Test
+	public void testMapMessageConvertingBothWaysJava() throws Exception {
+		Message<String> outMessage = MessageBuilder.withPayload("foo")
+				.setHeader("bar", "baz")
+				.build();
+		MapMessageConverter converter = new MapMessageConverter();
+		converter.setHeaderNames("bar");
+		MessageConvertingTcpMessageMapper mapper = new MessageConvertingTcpMessageMapper(converter);
+		Map<?, ?> map = (Map<?, ?>) mapper.fromMessage(outMessage);
+		DefaultSerializer serializer = new DefaultSerializer();
+		ByteArrayOutputStream baos = new ByteArrayOutputStream();
+		serializer.serialize(map, baos);
+
+		DefaultDeserializer deserializer = new DefaultDeserializer();
+		map = (Map<?, ?>) deserializer.deserialize(new ByteArrayInputStream(baos.toByteArray()));
+		TcpConnection connection = mock(TcpConnection.class);
+		when(connection.getPayload()).thenReturn(map);
+		when(connection.getHostName()).thenReturn("someHost");
+		when(connection.getHostAddress()).thenReturn("1.1.1.1");
+		when(connection.getPort()).thenReturn(1234);
+		when(connection.getConnectionId()).thenReturn("someId");
+		Message<?> message = mapper.toMessage(connection);
+		assertEquals("foo", message.getPayload());
+		assertEquals("baz", message.getHeaders().get("bar"));
+		assertEquals("someHost", message.getHeaders().get(IpHeaders.HOSTNAME));
+		assertEquals("1.1.1.1", message.getHeaders().get(IpHeaders.IP_ADDRESS));
+		assertEquals(1234, message.getHeaders().get(IpHeaders.REMOTE_PORT));
+		assertEquals("someId", message.getHeaders().get(IpHeaders.CONNECTION_ID));
+	}
+
+}

spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNetConnectionTests.java

@@ -21,19 +21,30 @@ import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
+import java.io.ByteArrayOutputStream;
 import java.io.InputStream;
+import java.io.PipedInputStream;
+import java.io.PipedOutputStream;
 import java.net.Socket;
 import java.nio.channels.SocketChannel;
 import java.util.concurrent.atomic.AtomicReference;
 
 import org.apache.commons.logging.Log;
+
 import org.junit.Test;
 import org.mockito.Mockito;
 import org.mockito.invocation.InvocationOnMock;
 import org.mockito.stubbing.Answer;
+
 import org.springframework.beans.DirectFieldAccessor;
+import org.springframework.context.ApplicationEvent;
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.integration.Message;
 import org.springframework.integration.ip.tcp.connection.TcpNioConnection.ChannelInputStream;
 import org.springframework.integration.ip.tcp.serializer.ByteArrayStxEtxSerializer;
+import org.springframework.integration.ip.tcp.serializer.MapJsonSerializer;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.integration.support.converter.MapMessageConverter;
 import org.springframework.integration.test.util.TestUtils;
 
 /**
@@ -43,13 +54,18 @@ import org.springframework.integration.test.util.TestUtils;
  */
 public class TcpNetConnectionTests {
 
+	private final ApplicationEventPublisher nullPublisher = new ApplicationEventPublisher() {
+		public void publishEvent(ApplicationEvent event) {
+		}
+	};
+
 	@Test
 	public void testErrorLog() throws Exception {
 		Socket socket = mock(Socket.class);
 		InputStream stream = mock(InputStream.class);
 		when(socket.getInputStream()).thenReturn(stream);
 		when(stream.read()).thenReturn((int) 'x');
-		TcpNetConnection connection = new TcpNetConnection(socket, true, false, null, null);
+		TcpNetConnection connection = new TcpNetConnection(socket, true, false, nullPublisher, null);
 		connection.setDeserializer(new ByteArrayStxEtxSerializer());
 		final AtomicReference<Object> log = new AtomicReference<Object>();
 		Log logger = mock(Log.class);
@@ -76,10 +92,55 @@ public class TcpNetConnectionTests {
 		SocketChannel socketChannel = mock(SocketChannel.class);
 		Socket socket = mock(Socket.class);
 		when(socketChannel.socket()).thenReturn(socket);
-		TcpNioConnection connection = new TcpNioConnection(socketChannel, true, false, null, null);
+		TcpNioConnection connection = new TcpNioConnection(socketChannel, true, false, nullPublisher, null);
 		ChannelInputStream inputStream = TestUtils.getPropertyValue(connection, "channelInputStream", ChannelInputStream.class);
 		inputStream.write(new byte[] {(byte) 0x80}, 1);
 		assertEquals(0x80, inputStream.read());
 	}
 
+	@Test
+	public void transferHeaders() throws Exception {
+		Socket inSocket = mock(Socket.class);
+		PipedInputStream pipe = new PipedInputStream();
+		when(inSocket.getInputStream()).thenReturn(pipe);
+
+		TcpConnectionSupport inboundConnection = new TcpNetConnection(inSocket, true, false, nullPublisher, null);
+		inboundConnection.setDeserializer(new MapJsonSerializer());
+		MapMessageConverter inConverter = new MapMessageConverter();
+		MessageConvertingTcpMessageMapper inMapper = new MessageConvertingTcpMessageMapper(inConverter);
+		inboundConnection.setMapper(inMapper);
+
+		ByteArrayOutputStream baos = new ByteArrayOutputStream();
+		Socket outSocket = mock(Socket.class);
+		TcpNetConnection outboundConnection = new TcpNetConnection(outSocket, true, false, nullPublisher, null);
+		when(outSocket.getOutputStream()).thenReturn(baos);
+
+		MapMessageConverter outConverter = new MapMessageConverter();
+		outConverter.setHeaderNames("bar");
+		MessageConvertingTcpMessageMapper outMapper = new MessageConvertingTcpMessageMapper(outConverter);
+		outboundConnection.setMapper(outMapper);
+		outboundConnection.setSerializer(new MapJsonSerializer());
+
+		Message<String> message = MessageBuilder.withPayload("foo")
+				.setHeader("bar", "baz")
+				.build();
+		outboundConnection.send(message);
+		PipedOutputStream out = new PipedOutputStream(pipe);
+		out.write(baos.toByteArray());
+		out.close();
+
+		final AtomicReference<Message<?>> inboundMessage = new AtomicReference<Message<?>>();
+		TcpListener listener = new TcpListener() {
+
+			public boolean onMessage(Message<?> message) {
+				inboundMessage.set(message);
+				return false;
+			}
+		};
+		inboundConnection.registerListener(listener);
+		inboundConnection.run();
+		assertNotNull(inboundMessage.get());
+		assertEquals("foo", inboundMessage.get().getPayload());
+		assertEquals("baz", inboundMessage.get().getHeaders().get("bar"));
+	}
 }

spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2012 the original author or authors.
+ * Copyright 2002-2013 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.
@@ -17,12 +17,15 @@
 package org.springframework.integration.ip.tcp.connection;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
+import static org.mockito.Matchers.any;
 import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
+import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.lang.reflect.Field;
@@ -46,18 +49,23 @@ import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
-
+import java.util.concurrent.atomic.AtomicReference;
 import javax.net.ServerSocketFactory;
 
 import org.junit.Test;
 import org.mockito.Mockito;
 import org.mockito.invocation.InvocationOnMock;
 import org.mockito.stubbing.Answer;
+
 import org.springframework.beans.DirectFieldAccessor;
+import org.springframework.context.ApplicationEvent;
+import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.integration.Message;
 import org.springframework.integration.ip.tcp.connection.TcpNioConnection.ChannelInputStream;
 import org.springframework.integration.ip.tcp.serializer.ByteArrayCrLfSerializer;
+import org.springframework.integration.ip.tcp.serializer.MapJsonSerializer;
 import org.springframework.integration.support.MessageBuilder;
+import org.springframework.integration.support.converter.MapMessageConverter;
 import org.springframework.integration.test.util.SocketUtils;
 import org.springframework.integration.test.util.TestUtils;
 import org.springframework.util.ReflectionUtils;
@@ -72,6 +80,11 @@ import org.springframework.util.ReflectionUtils.FieldFilter;
  */
 public class TcpNioConnectionTests {
 
+	private final ApplicationEventPublisher nullPublisher = new ApplicationEventPublisher() {
+		public void publishEvent(ApplicationEvent event) {
+		}
+	};
+
 	@Test
 	public void testWriteTimeout() throws Exception {
 		final int port = SocketUtils.findAvailableServerSocket();
@@ -442,6 +455,70 @@ public class TcpNioConnectionTests {
 		assertEquals("foo\u0000", new String(out));
 	}
 
+	@Test
+	public void transferHeaders() throws Exception {
+		Socket inSocket = mock(Socket.class);
+		SocketChannel inChannel = mock(SocketChannel.class);
+		when(inChannel.socket()).thenReturn(inSocket);
+
+		TcpNioConnection inboundConnection = new TcpNioConnection(inChannel, true, false, nullPublisher, null);
+		inboundConnection.setDeserializer(new MapJsonSerializer());
+		MapMessageConverter inConverter = new MapMessageConverter();
+		MessageConvertingTcpMessageMapper inMapper = new MessageConvertingTcpMessageMapper(inConverter);
+		inboundConnection.setMapper(inMapper);
+		final ByteArrayOutputStream written = new ByteArrayOutputStream();
+		doAnswer(new Answer<Integer>() {
+			public Integer answer(InvocationOnMock invocation) throws Throwable {
+				ByteBuffer buff = (ByteBuffer) invocation.getArguments()[0];
+				byte[] bytes = written.toByteArray();
+				buff.put(bytes);
+				return bytes.length;
+			}
+		}).when(inChannel).read(any(ByteBuffer.class));
+
+		Socket outSocket = mock(Socket.class);
+		SocketChannel outChannel = mock(SocketChannel.class);
+		when(outChannel.socket()).thenReturn(outSocket);
+		TcpNioConnection outboundConnection = new TcpNioConnection(outChannel, true, false, nullPublisher, null);
+		doAnswer(new Answer<Object>() {
+			public Object answer(InvocationOnMock invocation) throws Throwable {
+				ByteBuffer buff = (ByteBuffer) invocation.getArguments()[0];
+				byte[] bytes = new byte[buff.limit()];
+				buff.get(bytes);
+				written.write(bytes);
+				return null;
+			}
+		}).when(outChannel).write(any(ByteBuffer.class));
+
+		MapMessageConverter outConverter = new MapMessageConverter();
+		outConverter.setHeaderNames("bar");
+		MessageConvertingTcpMessageMapper outMapper = new MessageConvertingTcpMessageMapper(outConverter);
+		outboundConnection.setMapper(outMapper);
+		outboundConnection.setSerializer(new MapJsonSerializer());
+
+		Message<String> message = MessageBuilder.withPayload("foo")
+				.setHeader("bar", "baz")
+				.build();
+		outboundConnection.send(message);
+
+		final AtomicReference<Message<?>> inboundMessage = new AtomicReference<Message<?>>();
+		final CountDownLatch latch = new CountDownLatch(1);
+		TcpListener listener = new TcpListener() {
+
+			public boolean onMessage(Message<?> message) {
+				inboundMessage.set(message);
+				latch.countDown();
+				return false;
+			}
+		};
+		inboundConnection.registerListener(listener);
+		inboundConnection.readPacket();
+		assertTrue(latch.await(10, TimeUnit.SECONDS));
+		assertNotNull(inboundMessage.get());
+		assertEquals("foo", inboundMessage.get().getPayload());
+		assertEquals("baz", inboundMessage.get().getHeaders().get("bar"));
+	}
+
 	private void readFully(InputStream is, byte[] buff) throws IOException {
 		for (int i = 0; i < buff.length; i++) {
 			buff[i] = (byte) is.read();

spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/serializer/MapJsonSerializerTests.java

@@ -0,0 +1,43 @@
+/*
+ * Copyright 2002-2013 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.integration.ip.tcp.serializer;
+
+import static org.junit.Assert.assertNotNull;
+
+import java.io.ByteArrayInputStream;
+import java.util.Map;
+
+import org.junit.Test;
+
+/**
+ * @author Gary Russell
+ * @since 3.0
+ *
+ */
+public class MapJsonSerializerTests {
+
+	@Test
+	public void multi() throws Exception {
+		String json = "{\"headers\":{\"bar\":\"baz\"},\"payload\":\"foo\"}\n";
+		String twoJson = json + json;
+		MapJsonSerializer deserializer = new MapJsonSerializer();
+		ByteArrayInputStream bais = new ByteArrayInputStream(twoJson.getBytes("UTF-8"));
+		Map<?, ?> map = deserializer.deserialize(bais);
+		assertNotNull(map);
+		map = deserializer.deserialize(bais);
+		assertNotNull(map);
+	}
+}

src/reference/docbook/ip.xml

@@ -159,9 +159,10 @@
        <para>
         This is the extent of message correlation performed when sharing connection
         factories between inbound and outbound adapters. Such sharing allows for
-        asynchronous two-way communication over TCP. Only payload information is
+        asynchronous two-way communication over TCP. By default, only payload information is
         transferred using TCP; therefore any message correlation must be performed
-        by downstream components such as aggregators or other endpoints.
+        by downstream components such as aggregators or other endpoints. Support for
+        transferring selected headers was introduced in version 3.0.
         For more information refer to <xref linkend="ip-correlation" />.
        </para>
       </tip>
@@ -225,7 +226,17 @@
       telnet as a client, for example.
      </para>
      <para>
-      The <classname>ByteArrayStxEtxSerializer</classname>,
+      The <classname>ByteArraySingleTerminatorSerializer</classname>,
+      converts a byte array to a stream of bytes followed by a single termination
+      character (default 0x00).
+     </para>
+     <para>
+      The <classname>ByteArrayLfSerializer</classname>,
+      converts a byte array to a stream of bytes followed by a single linefeed
+      character (0x0a).
+     </para>
+     <para>
+		 The <classname>ByteArrayStxEtxSerializer</classname>,
       converts a byte array to a stream of bytes preceded by an STX (0x02) and
       followed by an ETX (0x03).
      </para>
@@ -276,6 +287,21 @@
       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.
      </para>
+     <para>
+      The <classname>MapJsonSerializer</classname> uses a Jackson
+      <classname>ObjectMapper</classname> to convert between a <interfacename>Map</interfacename>
+      and JSON. This can be used in conjunction with a <classname>MessageConvertingTcpMessageMapper
+      </classname> and a <classname>MapMessageConverter</classname>
+      to transfer selected headers and the payload in a JSON format.
+      <note>
+        The Jackson <classname>ObjectMapper</classname> cannot demarcate messages in the stream.
+        Therefore, the <classname>MapJsonSerializer</classname> needs to delegate to another
+        (de)serializer to handle message demarcation. By default, a
+        <classname>ByteArrayLfSerializer</classname> is used, resulting in messages with the
+        format <code>&lt;json&gt;&lt;LF&gt;</code> on the wire, but you can configure it to
+        use others instead.
+      </note>
+     </para>
      <para>
       The final standard serializer is
       <classname>org.springframework.core.serializer.DefaultSerializer</classname> which can be
@@ -677,8 +703,12 @@
     <para>
      One goal of the IP Endpoints is to provide communication with systems other
      than another Spring Integration application. For this reason, only
-     message payloads are sent
-     and received. No message correlation is provided by the framework,
+     message payloads are sent and received, by default. Since 3.0, headers can
+     be transferred, using JSON, Java serialization, or with custom
+     <interfacename>Serializer</interfacename>s and
+     <interfacename>Deserializer</interfacename>s; see <xref linkend="ip-headers"/>
+     for more information.
+     No message correlation is provided by the framework,
      except when using the gateways, or collaborating channel adapters on the
      server side. In the paragraphs below we discuss the various
      correlation techniques available to applications. In most cases, this
@@ -799,6 +829,98 @@
         </note>
     </para>
    </section>
+   <section id="ip-headers">
+     <title>Transferring Headers</title>
+     <para>
+       TCP is a streaming protocol; <interfacename>Serializers</interfacename> and
+       <interfacename>Deserializers</interfacename> are used to demarcate messages
+       within the stream. Prior to 3.0, only message payloads (String or byte[])
+       could be transferred over
+       TCP. Beginning with 3.0, you can now transfer selected headers as well as the
+       payload. It is important to understand, though, that "live" objects, such
+       as the <code>replyChannel</code> header cannot be serialized.
+     </para>
+     <para>
+       Sending header information over TCP requires some additional configuration.
+     </para>
+     <para>
+       The first step is to provide the <classname>ConnectionFactory</classname>
+       with a <classname>MessageConvertingTcpMessageMapper</classname> using the
+       <code>mapper</code> attribute. This mapper delegates to any
+       <interfacename>MessageConverter</interfacename> implementation to convert
+       the message to/from some object that can be (de)serialized by the
+       configured <code>serializer</code> and <code>deserializer</code>.
+     </para>
+     <para>
+       A <classname>MapMessageConverter</classname> is provided, which allows the
+       specification of a list of headers that will be added to a
+       <interfacename>Map</interfacename> object, along with the payload.
+       The generated Map has two entries: <code>payload</code> and
+       <code>headers</code>. The <code>headers</code> entry is itself a
+       <interfacename>Map</interfacename> containing the selected headers.
+     </para>
+     <para>
+       The second step is to provide a (de)serializer that can convert between
+       a <interfacename>Map</interfacename> and some wire format. This can
+       be a custom <interfacename>(de)Serializer</interfacename>, which would
+       typically be needed if the peer system is not a Spring Integration
+       application.
+     </para>
+     <para>
+       A <classname>MapJsonSerializer</classname> is provided that will
+       convert a Map to/from JSON. This uses a Spring Integration
+       <classname>JsonObjectMapper</classname> to perform this function. You
+       can provide a custom <classname>JsonObjectMapper</classname> if needed.
+       By default, the serializer inserts a linefeed
+       <code>0x0a</code> character between objects.
+       See the JavaDocs for more information.
+     </para>
+     <note>
+       At the time of writing, the <classname>JsonObjectMapper</classname>
+       uses whichever version of <code>Jackson</code> is on the classpath.
+     </note>
+     <para>
+       You can also use standard Java serialization of the Map, using
+       the <classname>DefaultSerializer</classname> and
+       <classname>DefaultDeserializer</classname>.
+     </para>
+     <para>
+       The following example shows the configuration of a connection
+       factory that transfers the <code>correlationId</code>,
+       <code>sequenceNumber</code>, and <code>sequenceSize</code>
+       headers using JSON.
+     </para>
+     <programlisting language="xml"><![CDATA[<int-ip:tcp-connection-factory id="client"
+    type="client"
+    host="localhost"
+    port="12345"
+    mapper="mapper"
+    serializer="jsonSerializer"
+    deserializer="jsonSerializer"/>
+
+<bean id="mapper"
+      class="o.sf.integration.ip.tcp.connection.MessageConvertingTcpMessageMapper">
+    <constructor-arg name="messageConverter">
+        <bean class="o.sf.integration.support.converter.MapMessageConverter">
+            <property name="headerNames">
+                <list>
+                    <value>correlationId</value>
+                    <value>sequenceNumber</value>
+                    <value>sequenceSize</value>
+                </list>
+            </property>
+        </bean>
+    </constructor-arg>
+</bean>
+
+<bean id="jsonSerializer" class="o.sf.integration.ip.tcp.serializer.MapJsonSerializer" />
+     ]]></programlisting>
+     <para>
+       A message sent with the above configuration, with payload 'foo' would appear on the wire like so:
+     </para>
+     <programlisting language="xml"><![CDATA[
+{"headers":{"correlationId":"bar","sequenceSize":5,"sequenceNumber":1},"payload":"foo"}]]></programlisting>
+   </section>
   </section>
   <section id="note_nio">
    <title>A Note About NIO</title>

src/reference/docbook/whats-new.xml

@@ -259,5 +259,18 @@
 				For more information see <xref linkend="message-id-generation"/>.
 			</para>
 		</section>
+		<section id="3.0-tcp-headers">
+			<title>Message Headers and TCP</title>
+			<para>
+				The TCP connection factories now enable the configuration of a flexible mechanism to
+				transfer selected headers (as well as the payload) over TCP. A new
+				<classname>TcpMessageMapper</classname>
+				enables the selection of the headers, and an appropriate (de)serializer needs to be
+				configured to write the resulting <interfacename>Map</interfacename> to the
+				TCP stream. A <classname>MapJsonSerializer</classname> is provided as a convenient
+				mechanism to transfer headers and payload over TCP.
+				For more information see <xref linkend="ip-headers"/>.
+			</para>
+		</section>
 	</section>
 </chapter>