INT-3779: CodecMessageConverter and Docs

JIRA: https://jira.spring.io/browse/INT-3779

Fix timeouts for the `EnableIntegrationTests`: https://build.spring.io/browse/INT-B41-385

spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/FileKryoRegistrar.java

@@ -25,18 +25,17 @@ import com.esotericsoftware.kryo.Registration;
  * A {@link KryoRegistrar} used to validateRegistration a File serializer.
  *
  * @author David Turanski
+ * @author Gary Russell
  * @since 4.2
  */
 public class FileKryoRegistrar extends AbstractKryoRegistrar {
 
-	private final static int DEFAULT_REGISTRATION_ID = 40;
-
 	private final int registrationId;
 
 	private final FileSerializer fileSerializer = new FileSerializer();
 
 	public FileKryoRegistrar() {
-		this.registrationId = DEFAULT_REGISTRATION_ID;
+		this.registrationId = RegistrationIds.DEFAULT_FILE_REGISTRATION_ID;
 	}
 
 	/**

spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/MessageCodec.java

@@ -0,0 +1,43 @@
+/*
+ * Copyright 2015 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.codec.kryo;
+
+/**
+ * {@link PojoCodec} configured to encode/decode {@code Message<?>}s.
+ * @author Gary Russell
+ * @since 4.2
+ *
+ */
+public class MessageCodec extends PojoCodec {
+
+	/**
+	 * Construct an instance using the default registration ids for messsage
+	 * headers.
+	 */
+	public MessageCodec() {
+		super(new MessageKryoRegistrar());
+	}
+
+	/**
+	 * Construct an instance using a custom registrar - usually used if different
+	 * registration ids are required for message headers.
+	 * @param registrar the registrar.
+	 */
+	public MessageCodec(MessageKryoRegistrar registrar) {
+		super(registrar);
+	}
+
+}

spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/MessageHeadersSerializer.java

@@ -0,0 +1,52 @@
+/*
+ * Copyright 2015 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.codec.kryo;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.springframework.messaging.MessageHeaders;
+
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.Serializer;
+import com.esotericsoftware.kryo.io.Input;
+import com.esotericsoftware.kryo.io.Output;
+
+/**
+ * Kryo Serializer for {@link MessageHeaders}.
+ * @author David Turanski
+ * @since 4.2
+ */
+class MessageHeadersSerializer extends Serializer<MessageHeaders> {
+	@Override
+	public void write(Kryo kryo, Output output, MessageHeaders headers) {
+		HashMap<String, Object> map = new HashMap<String, Object>();
+		for (Map.Entry<String, Object> entry : headers.entrySet()) {
+			if (entry.getValue() != null) {
+				map.put(entry.getKey(), entry.getValue());
+			}
+		}
+		kryo.writeObject(output, map);
+	}
+
+	@Override
+	public MessageHeaders read(Kryo kryo, Input input, Class<MessageHeaders> type) {
+		@SuppressWarnings("unchecked")
+		Map<String, Object> headers = kryo.readObject(input, HashMap.class);
+		return new MessageHeaders(headers);
+	}
+
+}

spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/MessageKryoRegistrar.java

@@ -0,0 +1,64 @@
+/*
+ * Copyright 2015 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.codec.kryo;
+
+import java.util.Arrays;
+import java.util.List;
+
+import org.springframework.integration.support.MutableMessageHeaders;
+import org.springframework.messaging.MessageHeaders;
+
+import com.esotericsoftware.kryo.Registration;
+
+/**
+ * Registers common MessageHeader types and Serializers.
+ * @author David Turanski
+ * @author Gary Russell
+ * @since 4.2
+ */
+public class MessageKryoRegistrar extends AbstractKryoRegistrar {
+
+	private volatile int messageHeadersRegistrationId = RegistrationIds.DEFAULT_MESSAGEHEADERS_ID;
+
+	private volatile int mutableMessageHeadersRegistrationId = RegistrationIds.DEFAULT_MUTABLE_MESSAGEHEADERS_ID;
+
+	/**
+	 * Set the registration id for {@code MessageHeaders}.
+	 * @param messageHeadersRegistrationId the id, default 41.
+	 */
+	public void setMessageHeadersRegistrationId(int messageHeadersRegistrationId) {
+		this.messageHeadersRegistrationId = messageHeadersRegistrationId;
+	}
+
+	/**
+	 * Set the registration id for {@code MutableMessageHeaders}.
+	 * @param mutableMessageHeadersRegistrationId the id, default 42.
+	 */
+	public void setMutableMessageHeadersRegistrationId(int mutableMessageHeadersRegistrationId) {
+		this.mutableMessageHeadersRegistrationId = mutableMessageHeadersRegistrationId;
+	}
+
+
+	@Override
+	public List<Registration> getRegistrations() {
+		return Arrays.asList(
+				new Registration(MessageHeaders.class, new MessageHeadersSerializer(),
+						this.messageHeadersRegistrationId),
+				new Registration(MutableMessageHeaders.class, new MutableMessageHeadersSerializer(),
+						this.mutableMessageHeadersRegistrationId));
+	}
+
+}

spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/MutableMessageHeadersSerializer.java

@@ -0,0 +1,41 @@
+/*
+ * Copyright 2015 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.codec.kryo;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.springframework.integration.support.MutableMessageHeaders;
+import org.springframework.messaging.MessageHeaders;
+
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.io.Input;
+
+/**
+ * Kryo Serializer for {@link MutableMessageHeaders}.
+ * @author David Turanski
+ * @since 4.2
+ */
+class MutableMessageHeadersSerializer extends MessageHeadersSerializer {
+
+	@Override
+	public MessageHeaders read(Kryo kryo, Input input, Class<MessageHeaders> type) {
+		@SuppressWarnings("unchecked")
+		Map<String, Object> headers = kryo.readObject(input, HashMap.class);
+		return new MutableMessageHeaders(headers);
+	}
+
+}

spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/RegistrationIds.java

@@ -0,0 +1,35 @@
+/*
+ * Copyright 2015 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.codec.kryo;
+
+/**
+ * Default registration ids for serializers provided by the framework.
+ *
+ * @author Gary Russell
+ * @since 4.2
+ *
+ */
+public final class RegistrationIds {
+
+	public final static int DEFAULT_FILE_REGISTRATION_ID = 40;
+
+	public static final int DEFAULT_MESSAGEHEADERS_ID = 41;
+
+	public static final int DEFAULT_MUTABLE_MESSAGEHEADERS_ID = 42;
+
+	private RegistrationIds() {}
+
+}

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

@@ -37,10 +37,11 @@ import org.springframework.util.ObjectUtils;
  * @author Gary Russell
  * @author Artem Bilan
  * @author Stuart Williams
+ * @author David Turanski
  * @since 4.0
  *
  */
-class MutableMessage<T> implements Message<T>, Serializable {
+public class MutableMessage<T> implements Message<T>, Serializable {
 
 	private static final long serialVersionUID = -636635024258737500L;
 
@@ -48,11 +49,11 @@ class MutableMessage<T> implements Message<T>, Serializable {
 
 	private final MutableMessageHeaders headers;
 
-	MutableMessage(T payload) {
+	public MutableMessage(T payload) {
 		this(payload, null);
 	}
 
-	MutableMessage(T payload, Map<String, Object> headers) {
+	public MutableMessage(T payload, Map<String, Object> headers) {
 		Assert.notNull(payload, "payload must not be null");
 		this.payload = payload;
 

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

@@ -26,13 +26,14 @@ import org.springframework.messaging.MessageHeaders;
  * header map.
  *
  * @author Stuart Williams
+ * @author David Turanski
  * @since 4.2
  */
-class MutableMessageHeaders extends MessageHeaders {
+public class MutableMessageHeaders extends MessageHeaders {
 
 	private static final long serialVersionUID = 3084692953798643018L;
 
-	MutableMessageHeaders(Map<String, Object> headers) {
+	public MutableMessageHeaders(Map<String, Object> headers) {
 		super(headers);
 	}
 

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

@@ -0,0 +1,77 @@
+/*
+ * Copyright 2015 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.io.IOException;
+
+import org.springframework.integration.codec.Codec;
+import org.springframework.integration.context.IntegrationObjectSupport;
+import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.messaging.MessagingException;
+import org.springframework.messaging.converter.MessageConverter;
+import org.springframework.messaging.support.GenericMessage;
+import org.springframework.util.Assert;
+
+/**
+ * A {@link MessageConverter} that delegates to a {@link Codec} to convert
+ *
+ * @author Gary Russell
+ * @since 4.2
+ *
+ */
+public class CodecMessageConverter extends IntegrationObjectSupport implements MessageConverter {
+
+	private final Codec codec;
+
+	private final Class<?> messageClass;
+
+	public CodecMessageConverter(Codec codec) {
+		this.codec = codec;
+		this.messageClass = GenericMessage.class;
+	}
+
+	@Override
+	public Object fromMessage(Message<?> message, Class<?> targetClass) {
+		try {
+			return this.codec.encode(message);
+		}
+		catch (IOException e) {
+			throw new MessagingException(message, e);
+		}
+	}
+
+	@Override
+	public Message<?> toMessage(Object payload, MessageHeaders headers) {
+		Assert.isInstanceOf(byte[].class, payload);
+		try {
+			Message<?> decoded = (Message<?>) this.codec.decode((byte[]) payload, messageClass);
+			if (headers != null) {
+				AbstractIntegrationMessageBuilder<?> builder = getMessageBuilderFactory().fromMessage(decoded);
+				builder.copyHeaders(headers);
+				return builder.build();
+			}
+			else {
+				return decoded;
+			}
+		}
+		catch (IOException e) {
+			throw new MessagingException("Failed to decode", e);
+		}
+	}
+
+}

spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.java

@@ -1262,7 +1262,7 @@ public class EnableIntegrationTests {
 	@Target({ElementType.TYPE, ElementType.ANNOTATION_TYPE})
 	@Retention(RetentionPolicy.RUNTIME)
 	@MessagingGateway(defaultRequestChannel = "gatewayChannel", reactorEnvironment = "reactorEnv",
-			defaultRequestTimeout="${default.request.timeout:23}", defaultReplyTimeout="#{34}",
+			defaultRequestTimeout="${default.request.timeout:12300}", defaultReplyTimeout="#{13400}",
 			defaultHeaders = @GatewayHeader(name = "foo", value = "FOO"))
 	public @interface TestMessagingGateway {
 

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

@@ -24,21 +24,28 @@ import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
 import java.net.Socket;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.Map;
 
 import javax.net.SocketFactory;
 import javax.net.ssl.SSLSession;
 
+import org.junit.Before;
 import org.junit.Test;
 
 import org.springframework.core.serializer.DefaultDeserializer;
 import org.springframework.core.serializer.DefaultSerializer;
 import org.springframework.integration.IntegrationMessageHeaderAccessor;
+import org.springframework.integration.codec.Codec;
+import org.springframework.integration.codec.CompositeCodec;
+import org.springframework.integration.codec.kryo.MessageCodec;
 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.CodecMessageConverter;
 import org.springframework.integration.support.converter.MapMessageConverter;
 import org.springframework.messaging.Message;
+import org.springframework.messaging.converter.MessageConverter;
 
 /**
  * @author Gary Russell
@@ -49,6 +56,14 @@ public class TcpMessageMapperTests {
 
 	private static final String TEST_PAYLOAD = "abcdefghijkl";
 
+	private Codec codec;
+
+	@Before
+	public void setup() {
+		Map<Class<?>, Codec> codecs = new HashMap<Class<?>, Codec>();
+		this.codec = new CompositeCodec(codecs, new MessageCodec());
+	}
+
 	@Test
 	public void testToMessage() throws Exception {
 
@@ -323,4 +338,29 @@ public class TcpMessageMapperTests {
 		assertEquals(1234, message.getHeaders().get(IpHeaders.REMOTE_PORT));
 		assertEquals("someId", message.getHeaders().get(IpHeaders.CONNECTION_ID));
 	}
+
+	@Test
+	public void testCodecMessageConvertingBothWaysJava() throws Exception {
+		Message<String> outMessage = MessageBuilder.withPayload("foo")
+				.setHeader("bar", "baz")
+				.build();
+		MessageConverter converter = new CodecMessageConverter(this.codec);
+		MessageConvertingTcpMessageMapper mapper = new MessageConvertingTcpMessageMapper(converter);
+		byte[] bytes = (byte[]) mapper.fromMessage(outMessage);
+
+		TcpConnection connection = mock(TcpConnection.class);
+		when(connection.getPayload()).thenReturn(bytes);
+		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));
+	}
+
 }

src/reference/asciidoc/codec.adoc

@@ -0,0 +1,139 @@
+[[codec]]
+=== Codec
+
+==== Introduction
+
+Spring Integration _version 4.2_ introduces the `Codec` abstraction.
+Codecs are used to encode/decode objects to/from `byte[]`.
+They are an alternative to Java Serialization.
+One advantage is, typically, objects do not have to implement `Serializable`.
+One implementation, using https://github.com/EsotericSoftware/kryo[Kryo] for serialization, is provided but you
+can provide your own implementation for use in any of these components:
+
+* `EncodingPayloadTransformer`
+* `DecodingTransformer`
+* `CodecMessageConverter`
+
+See their JavaDocs for more information.
+
+==== EncodingPayloadTransformer
+
+This transformer encodes the payload to a `byte[]` using the codec.
+It does not affect message headers.
+
+==== DecodingTransformer
+
+This transformer decodes a `byte[]` using the codec; it needs to be configured with the Class to which the object
+should be decoded (or an expression that resolves to a Class).
+If the resulting object is a `Message<?>`, inbound headers will not be retained.
+
+==== CodecMessageConverter
+
+Certain endpoints (e.g. TCP, Redis) have no concept of message headers; they support the use of a
+`MessageConverter` and the `CodecMessageConverter` can be used to convert a message to/from a `byte[]` for
+transmission.
+
+==== Kryo
+
+Currently, this is the only implementation of `Codec`.
+There are two `Codec` s - `PojoCodec` which can be used in the transformers and `MessageCodec` which can be used
+in the `CodecMessageConverter`.
+
+Several custom serializers are provided by the framework:
+
+* `FileSerializer`
+* `MessageHeadersSerializer`
+* `MutableMessageHeadersSerializer`
+
+The first can be used with the `PojoCodec`, by initializing it with the `FileKryoRegistrar`.
+The second and third are used with the `MessageCodec`, which is initialized with the `MessageKryoRegistrar`.
+
+===== Customizing Kryo
+
+By default, Kryo delegates unknown Java types to its `FieldSerializer`.
+Kryo also registers default serializers for each primitive type along with `String`, `Collection` and `Map` serializers.
+`FieldSerializer` uses reflection to navigate the object graph. A more efficient approach is to implement a custom
+serializer that is aware of the object's structure and can directly serialize selected primitive fields:
+
+[source,java]
+----
+public class AddressSerializer extends Serializer<Address> {
+	@Override
+	public void write(Kryo kryo, Output output, Address address) {
+		output.writeString(address.getStreet());
+		output.writeString(address.getCity());
+		output.writeString(address.getCountry());
+	}
+
+	@Override
+	public Address read(Kryo kryo, Input input, Class<Address> type) {
+		return new Address(input.readString(),input.readString(),input.readString());
+	}
+}
+----
+
+The `Serializer` interface exposes `Kryo`, `Input`, and `Output` which provide
+complete control over which fields are included and other internal settings as
+described in the https://github.com/EsotericSoftware/kryo[documentation].
+
+NOTE: When registering your custom serializer, you need a registration ID.
+The registration IDs are arbitrary but in our case must be explicitly defined because each Kryo instance across the
+distributed application must use the same IDs.
+Kryo recommends small positive integers, and reserves a few ids (value < 10).
+Spring Integration currently defaults to using 40, 41 and 42 (for the file and message header serializers mentioned
+above); we recommend you start at, say 60, to allow for expansion in the framework.
+These framework defaults can be overridden by configuring the registrars mentioned above.
+
+====== Using a Custom Kryo Serializer
+
+If custom serialization is indicated, please consult the https://github.com/EsotericSoftware/kryo[Kryo] documentation
+since you will be using the native API.
+For an example, see the `MessageCodec`.
+
+====== Implementing KryoSerializable
+
+If you have write access to the domain object source code it may implement `KryoSerializable` as described
+https://github.com/EsotericSoftware/kryo#kryoserializable[here].
+In this case
+the class provides the serialization methods itself and no further configuration
+is required. This has the advantage of being much simpler to use
+with XD, however benchmarks have shown this is not quite as efficient as
+registering a custom serializer explicitly:
+
+[source,java]
+----
+public class Address implements KryoSerializable {
+	...
+
+	@Override
+	public void write(Kryo kryo, Output output) {
+		output.writeString(this.street);
+		output.writeString(this.city);
+		output.writeString(this.country);
+	}
+
+	@Override
+	public void read(Kryo kryo, Input input) {
+		this.street = input.readString();
+		this.city = input.readString();
+		this.country = input.readString();
+	}
+}
+----
+
+Note that this technique can also be used to wrap a serialization library other than Kryo.
+
+====== Using DefaultSerializer Annotation
+
+Kryo also provides an annotation as described https://github.com/EsotericSoftware/kryo#default-serializers[here].
+
+[source,java]
+----
+@DefaultSerializer(SomeClassSerializer.class)
+public class SomeClass {
+       // ...
+}
+----
+
+If you have write access to the domain object this may be a simpler alternative to specify a custom serializer.
+Note this does not register the class with an ID, so your mileage may vary.

src/reference/asciidoc/message-transformation.adoc

@@ -7,3 +7,5 @@ include::./transformer.adoc[]
 include::./content-enrichment.adoc[]
 
 include::./claim-check.adoc[]
+
+include::./codec.adoc[]

src/reference/asciidoc/transformer.adoc

@@ -384,3 +384,8 @@ It is a typical endpoint with input/output channels and a `header-names` attribu
 That attribute accepts the names of the header(s) (delimited by commas if there are multiple)
 that need to be removed.
 So, in the above example the headers named 'lastName' and 'state' will not be present on the outbound Message.
+
+
+==== Codec-Based Transformers
+
+See <<codec>>.

src/reference/asciidoc/whats-new.adoc

@@ -68,6 +68,15 @@ See <<barrier>> for more information.
 STOMP support has been added to the framework as _inbound_ and _outbound_ channel adapters pair.
 See <<stomp>> for more information.
 
+[[x4.2-codec]]
+==== Codec
+A new `Codec` abstraction has been introduced, to encode/decode objects to/from `byte[]`.
+An implementation that uses Kryo is provided.
+Codec-based transformers and message converters are also provided.
+
+See <<codec>> for more information.
+
+
 [[x4.2-general]]
 === General Changes