INT-4267: Add JSON (De)Serializers for Messaging

JIRA: https://jira.spring.io/browse/INT-4267
Fixes: spring-projects/spring-integration#2110

The documentation clearly point that we can simply use JSON (de)serialization
with the `RedisMessageStore`, but actually it isn't so easy

* Fix `MessageGroupMetadata`, `MessageHolder`, `MessageMetadata` for Jackson
deserialization compatibility
* Add `MessageHeaders`-based ctor to the `AdviceMessage`
* Add `MessageHeadersJacksonSerializer` to serialize `MessageHeaders`
to the `HashMap` for easier deserialization afterwards
* Add deserializer implementations for all `Message` types
* Add convenient `JsonObjectMapperProvider#jacksonMessageAwareMapper()`
factory method to build `ObjectMapper` supplied with mentioned above
(de)serializers

**Cherry-pick to 4.3.10 without `MessageHolder` and `MessageMetadata`**

Address PR comments and document the feature

Doc Polishing

Conflicts:
	spring-integration-core/src/main/java/org/springframework/integration/store/MessageHolder.java
	spring-integration-core/src/main/java/org/springframework/integration/store/MessageMetadata.java
Resolved.

spring-integration-core/src/main/java/org/springframework/integration/message/AdviceMessage.java

@@ -19,6 +19,7 @@ package org.springframework.integration.message;
 import java.util.Map;
 
 import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.support.GenericMessage;
 
 /**
@@ -49,6 +50,20 @@ public class AdviceMessage<T> extends GenericMessage<T> {
 		this.inputMessage = inputMessage;
 	}
 
+	/**
+	 * A constructor with the {@link MessageHeaders} instance to use.
+	 * <p><strong>Note:</strong> the given {@code MessageHeaders} instance is used
+	 * directly in the new message, i.e. it is not copied.
+	 * @param payload the message payload (never {@code null})
+	 * @param headers message headers
+	 * @param inputMessage the input message for advice.
+	 * @since 4.3.10
+	 */
+	public AdviceMessage(T payload, MessageHeaders headers, Message<?> inputMessage) {
+		super(payload, headers);
+		this.inputMessage = inputMessage;
+	}
+
 	public Message<?> getInputMessage() {
 		return this.inputMessage;
 	}

spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupMetadata.java

@@ -17,6 +17,7 @@
 package org.springframework.integration.store;
 
 import java.io.Serializable;
+import java.util.Collections;
 import java.util.Iterator;
 import java.util.LinkedList;
 import java.util.List;
@@ -31,17 +32,17 @@ import org.springframework.util.Assert;
  * @author Oleg Zhurakousky
  * @author Gary Russell
  * @author Artem Bilan
+ * @author Laszlo Szabo
+ *
  * @since 2.1
  */
 public class MessageGroupMetadata implements Serializable {
 
 	private static final long serialVersionUID = 1L;
 
-	private final Object groupId;
+	private List<UUID> messageIds = new LinkedList<UUID>();
 
-	private final List<UUID> messageIds = new LinkedList<UUID>();
-
-	private final long timestamp;
+	private long timestamp;
 
 	private volatile boolean complete;
 
@@ -49,9 +50,12 @@ public class MessageGroupMetadata implements Serializable {
 
 	private volatile int lastReleasedMessageSequenceNumber;
 
+	private MessageGroupMetadata() {
+		//For Jackson deserialization
+	}
+
 	public MessageGroupMetadata(MessageGroup messageGroup) {
 		Assert.notNull(messageGroup, "'messageGroup' must not be null");
-		this.groupId = messageGroup.getGroupId();
 		for (Message<?> message : messageGroup.getMessages()) {
 			this.messageIds.add(message.getHeaders().getId());
 		}
@@ -73,10 +77,6 @@ public class MessageGroupMetadata implements Serializable {
 		this.lastModified = lastModified;
 	}
 
-	public Object getGroupId() {
-		return this.groupId;
-	}
-
 	public Iterator<UUID> messageIdIterator() {
 		return this.messageIds.iterator();
 	}
@@ -92,6 +92,10 @@ public class MessageGroupMetadata implements Serializable {
 		return null;
 	}
 
+	public List<UUID> getMessageIds() {
+		return Collections.unmodifiableList(this.messageIds);
+	}
+
 	void complete() {
 		this.complete = true;
 	}

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

@@ -0,0 +1,52 @@
+/*
+ * Copyright 2017 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.json;
+
+import java.io.IOException;
+
+import org.springframework.integration.message.AdviceMessage;
+import org.springframework.integration.support.MutableMessageHeaders;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHeaders;
+
+import com.fasterxml.jackson.databind.DeserializationContext;
+import com.fasterxml.jackson.databind.JsonNode;
+
+/**
+ * The {@link MessageJacksonDeserializer} implementation for the {@link AdviceMessage}.
+ *
+ * @author Artem Bilan
+ *
+ * @since 4.3.10
+ */
+public class AdviceMessageJacksonDeserializer extends MessageJacksonDeserializer<AdviceMessage<?>> {
+
+	private static final long serialVersionUID = 1L;
+
+	@SuppressWarnings("unchecked")
+	public AdviceMessageJacksonDeserializer() {
+		super((Class<AdviceMessage<?>>) (Class<?>) AdviceMessage.class);
+	}
+
+	@Override
+	protected AdviceMessage<?> buildMessage(MutableMessageHeaders headers, Object payload, JsonNode root,
+			DeserializationContext ctxt) throws IOException {
+		Message<?> inputMessage = getMapper().readValue(root.get("inputMessage").traverse(), Message.class);
+		return new AdviceMessage<Object>(payload, (MessageHeaders) headers, inputMessage);
+	}
+
+}

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

@@ -0,0 +1,52 @@
+/*
+ * Copyright 2017 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.json;
+
+import java.io.IOException;
+
+import org.springframework.integration.support.MutableMessageHeaders;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.ErrorMessage;
+
+import com.fasterxml.jackson.databind.DeserializationContext;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.type.TypeFactory;
+
+/**
+ * The {@link MessageJacksonDeserializer} implementation for the {@link ErrorMessage}.
+ *
+ * @author Artem Bilan
+ * @author Gary Russell
+ *
+ * @since 4.3.10
+ */
+public class ErrorMessageJacksonDeserializer extends MessageJacksonDeserializer<ErrorMessage> {
+
+	private static final long serialVersionUID = 1L;
+
+	public ErrorMessageJacksonDeserializer() {
+		super(ErrorMessage.class);
+		setPayloadType(TypeFactory.defaultInstance().constructType(Throwable.class));
+	}
+
+	@Override
+	protected ErrorMessage buildMessage(MutableMessageHeaders headers, Object payload, JsonNode root,
+			DeserializationContext ctxt) throws IOException {
+		return new ErrorMessage((Throwable) payload, headers);
+	}
+
+}

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

@@ -0,0 +1,48 @@
+/*
+ * Copyright 2017 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.json;
+
+import java.io.IOException;
+
+import org.springframework.integration.support.MutableMessageHeaders;
+import org.springframework.messaging.support.GenericMessage;
+
+import com.fasterxml.jackson.databind.DeserializationContext;
+import com.fasterxml.jackson.databind.JsonNode;
+
+/**
+ * The {@link MessageJacksonDeserializer} implementation for the {@link GenericMessage}.
+ *
+ * @author Artem Bilan
+ *
+ * @since 4.3.10
+ */
+public class GenericMessageJacksonDeserializer extends MessageJacksonDeserializer<GenericMessage<?>> {
+
+	private static final long serialVersionUID = 1L;
+
+	@SuppressWarnings("unchecked")
+	public GenericMessageJacksonDeserializer() {
+		super((Class<GenericMessage<?>>) (Class<?>) GenericMessage.class);
+	}
+
+	@Override
+	protected GenericMessage<?> buildMessage(MutableMessageHeaders headers, Object payload, JsonNode root,
+			DeserializationContext ctxt) throws IOException {
+		return new GenericMessage<Object>(payload, headers);
+	}
+}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2017 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.
@@ -16,13 +16,24 @@
 
 package org.springframework.integration.support.json;
 
+import org.springframework.integration.message.AdviceMessage;
+import org.springframework.integration.support.MutableMessage;
+import org.springframework.messaging.support.ErrorMessage;
+import org.springframework.messaging.support.GenericMessage;
 import org.springframework.util.ClassUtils;
 
+import com.fasterxml.jackson.annotation.JsonTypeInfo;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.MapperFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.module.SimpleModule;
+
 /**
  * Utility methods for Jackson.
  *
  * @author Artem Bilan
  * @author Gary Russell
+ *
  * @since 3.0
  *
  */
@@ -50,4 +61,45 @@ public final class JacksonJsonUtils {
 		return jacksonPresent;
 	}
 
+	/**
+	 * Return an {@link ObjectMapper} if available,
+	 * supplied with Message specific serializers and deserializers.
+	 * Also configured to store typo info in the {@code @class} property.
+	 * @return the mapper.
+	 * @throws IllegalStateException if an implementation is not available.
+	 * @since 4.3.10
+	 */
+	public static ObjectMapper messagingAwareMapper() {
+		if (jackson2Present) {
+			ObjectMapper mapper = new ObjectMapper();
+			mapper.configure(MapperFeature.DEFAULT_VIEW_INCLUSION, false);
+			mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
+			mapper.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL, JsonTypeInfo.As.PROPERTY);
+
+			GenericMessageJacksonDeserializer genericMessageDeserializer = new GenericMessageJacksonDeserializer();
+			genericMessageDeserializer.setMapper(mapper);
+
+			ErrorMessageJacksonDeserializer errorMessageDeserializer = new ErrorMessageJacksonDeserializer();
+			errorMessageDeserializer.setMapper(mapper);
+
+			AdviceMessageJacksonDeserializer adviceMessageDeserializer = new AdviceMessageJacksonDeserializer();
+			adviceMessageDeserializer.setMapper(mapper);
+
+			MutableMessageJacksonDeserializer mutableMessageDeserializer = new MutableMessageJacksonDeserializer();
+			mutableMessageDeserializer.setMapper(mapper);
+
+			mapper.registerModule(new SimpleModule()
+					.addSerializer(new MessageHeadersJacksonSerializer())
+					.addDeserializer(GenericMessage.class, genericMessageDeserializer)
+					.addDeserializer(ErrorMessage.class, errorMessageDeserializer)
+					.addDeserializer(AdviceMessage.class, adviceMessageDeserializer)
+					.addDeserializer(MutableMessage.class, mutableMessageDeserializer)
+			);
+			return mapper;
+		}
+		else {
+			throw new IllegalStateException("No jackson-databind.jar is present in the classpath.");
+		}
+	}
+
 }

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

@@ -0,0 +1,59 @@
+/*
+ * Copyright 2017 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.json;
+
+import java.io.IOException;
+import java.util.HashMap;
+
+import org.springframework.messaging.MessageHeaders;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.databind.SerializerProvider;
+import com.fasterxml.jackson.databind.jsontype.TypeSerializer;
+import com.fasterxml.jackson.databind.ser.std.StdSerializer;
+
+/**
+ * A Jackson {@link StdSerializer} implementation to serialize {@link MessageHeaders}
+ * as a {@link HashMap}.
+ * <p>
+ * This technique is much reliable during deserialization, especially when the
+ * {@code typeId} property is used to store the type.
+ *
+ * @author Artem Bilan
+ *
+ * @since 4.3.10
+ */
+public class MessageHeadersJacksonSerializer extends StdSerializer<MessageHeaders> {
+
+	private static final long serialVersionUID = 1L;
+
+	public MessageHeadersJacksonSerializer() {
+		super(MessageHeaders.class);
+	}
+
+	@Override
+	public void serializeWithType(MessageHeaders value, JsonGenerator gen, SerializerProvider serializers,
+			TypeSerializer typeSer) throws IOException {
+		serialize(value, gen, serializers);
+	}
+
+	@Override
+	public void serialize(MessageHeaders value, JsonGenerator gen, SerializerProvider provider) throws IOException {
+		gen.writeObject(new HashMap<String, Object>(value));
+	}
+
+}

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

@@ -0,0 +1,80 @@
+/*
+ * Copyright 2017 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.json;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.springframework.integration.support.MutableMessageHeaders;
+import org.springframework.messaging.Message;
+import org.springframework.util.Assert;
+
+import com.fasterxml.jackson.annotation.JsonTypeInfo;
+import com.fasterxml.jackson.databind.DeserializationContext;
+import com.fasterxml.jackson.databind.JavaType;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.deser.std.StdNodeBasedDeserializer;
+import com.fasterxml.jackson.databind.type.TypeFactory;
+
+/**
+ * A Jackson {@link StdNodeBasedDeserializer} extension for {@link Message} implementations.
+ *
+ * @author Artem Bilan
+ *
+ * @since 4.3.10
+ */
+public abstract class MessageJacksonDeserializer<T extends Message<?>> extends StdNodeBasedDeserializer<T> {
+
+	private static final long serialVersionUID = 1L;
+
+	private JavaType payloadType = TypeFactory.defaultInstance().constructType(Object.class);
+
+	private ObjectMapper mapper = new ObjectMapper();
+
+	protected MessageJacksonDeserializer(Class<T> targetType) {
+		super(targetType);
+		this.mapper.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL, JsonTypeInfo.As.PROPERTY);
+	}
+
+	public void setMapper(ObjectMapper mapper) {
+		Assert.notNull(mapper, "'mapper' must not be null");
+		this.mapper = mapper;
+	}
+
+	protected final void setPayloadType(JavaType payloadType) {
+		Assert.notNull(payloadType, "'payloadType' must not be null");
+		this.payloadType = payloadType;
+	}
+
+	protected ObjectMapper getMapper() {
+		return this.mapper;
+	}
+
+	@Override
+	public T convert(JsonNode root, DeserializationContext ctxt) throws IOException {
+		Map<String, Object> headers = this.mapper.readValue(root.get("headers").traverse(),
+				TypeFactory.defaultInstance().constructMapType(HashMap.class, String.class, Object.class));
+		Object payload = this.mapper.readValue(root.get("payload").traverse(), this.payloadType);
+		return buildMessage(new MutableMessageHeaders(headers), payload, root, ctxt);
+	}
+
+	protected abstract T buildMessage(MutableMessageHeaders headers, Object payload, JsonNode root,
+			DeserializationContext ctxt) throws IOException;
+
+}

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

@@ -0,0 +1,49 @@
+/*
+ * Copyright 2017 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.json;
+
+import java.io.IOException;
+
+import org.springframework.integration.support.MutableMessage;
+import org.springframework.integration.support.MutableMessageHeaders;
+
+import com.fasterxml.jackson.databind.DeserializationContext;
+import com.fasterxml.jackson.databind.JsonNode;
+
+/**
+ * The {@link MessageJacksonDeserializer} implementation for the {@link MutableMessage}.
+ *
+ * @author Artem Bilan
+ *
+ * @since 4.3.10
+ */
+public class MutableMessageJacksonDeserializer extends MessageJacksonDeserializer<MutableMessage<?>> {
+
+	private static final long serialVersionUID = 1L;
+
+	@SuppressWarnings("unchecked")
+	public MutableMessageJacksonDeserializer() {
+		super((Class<MutableMessage<?>>) (Class<?>) MutableMessage.class);
+	}
+
+	@Override
+	protected MutableMessage<?> buildMessage(MutableMessageHeaders headers, Object payload, JsonNode root,
+			DeserializationContext ctxt) throws IOException {
+		return new MutableMessage<Object>(payload, headers);
+	}
+
+}

spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisMessageGroupStoreTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2007-2016 the original author or authors.
+ * Copyright 2007-2017 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.
@@ -16,16 +16,20 @@
 
 package org.springframework.integration.redis.store;
 
+import static org.hamcrest.Matchers.instanceOf;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNotSame;
 import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
 
 import java.util.ArrayList;
+import java.util.Date;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Properties;
+import java.util.UUID;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
@@ -38,18 +42,25 @@ import org.junit.Test;
 import org.springframework.context.support.ClassPathXmlApplicationContext;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
 import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.data.redis.serializer.GenericJackson2JsonRedisSerializer;
 import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.history.MessageHistory;
+import org.springframework.integration.message.AdviceMessage;
 import org.springframework.integration.redis.rules.RedisAvailable;
 import org.springframework.integration.redis.rules.RedisAvailableTests;
 import org.springframework.integration.store.MessageGroup;
 import org.springframework.integration.store.SimpleMessageGroup;
 import org.springframework.integration.support.MessageBuilder;
+import org.springframework.integration.support.MutableMessage;
+import org.springframework.integration.support.json.JacksonJsonUtils;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.support.ErrorMessage;
 import org.springframework.messaging.support.GenericMessage;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
+
 import junit.framework.AssertionFailedError;
 
 /**
@@ -421,4 +432,34 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests {
 		messageStore.removeMessageGroup("X");
 	}
 
+	@Test
+	@RedisAvailable
+	public void testJsonSerialization() throws Exception {
+		RedisConnectionFactory jcf = getConnectionFactoryForTest();
+		RedisMessageStore store = new RedisMessageStore(jcf);
+
+		ObjectMapper mapper = JacksonJsonUtils.messagingAwareMapper();
+
+		GenericJackson2JsonRedisSerializer serializer = new GenericJackson2JsonRedisSerializer(mapper);
+		store.setValueSerializer(serializer);
+
+		Message<?> genericMessage = new GenericMessage<>(new Date());
+		Message<?> mutableMessage = new MutableMessage<>(UUID.randomUUID());
+		Message<?> adviceMessage = new AdviceMessage<>("foo", genericMessage);
+		ErrorMessage errorMessage = new ErrorMessage(new RuntimeException("test exception"));
+
+		store.addMessagesToGroup(1, genericMessage, mutableMessage, adviceMessage, errorMessage);
+
+		MessageGroup messageGroup = store.getMessageGroup(1);
+		assertEquals(4, messageGroup.size());
+		List<Message<?>> messages = new ArrayList<>(messageGroup.getMessages());
+		assertEquals(genericMessage.getPayload(), messages.get(0).getPayload());
+		assertEquals(mutableMessage.getPayload(), messages.get(1).getPayload());
+		assertEquals(adviceMessage.getPayload(), messages.get(2).getPayload());
+		Message<?> errorMessageResult = messages.get(3);
+		assertThat(errorMessageResult, instanceOf(ErrorMessage.class));
+		assertEquals(errorMessage.getPayload().getMessage(),
+				((ErrorMessage) errorMessageResult).getPayload().getMessage());
+	}
+
 }

src/reference/asciidoc/redis.adoc

@@ -319,7 +319,7 @@ Handling these events using an `<int-event:inbound-channel-adapter/>` can be use
 
 As described in EIP, a http://www.eaipatterns.com/MessageStore.html[Message Store] allows you to persist Messages.
 This can be very useful when dealing with components that have a capability to buffer messages (_Aggregator, Resequencer_, etc.) if reliability is a concern.
-In Spring Integration, the MessageStore strategy also provides the foundation for thehttp://www.eaipatterns.com/StoreInLibrary.html[ClaimCheck] pattern, which is described in EIP as well.
+In Spring Integration, the MessageStore strategy also provides the foundation for the http://www.eaipatterns.com/StoreInLibrary.html[ClaimCheck] pattern, which is described in EIP as well.
 
 Spring Integration's Redis module provides the `RedisMessageStore`.
 
@@ -339,6 +339,21 @@ As you can see it is a simple bean configuration, and it expects a `RedisConnect
 By default the `RedisMessageStore` will use Java serialization to serialize the Message.
 However if you want to use a different serialization technique (e.g., JSON), you can provide your own serializer via the `valueSerializer` property of the `RedisMessageStore`.
 
+Starting with _version 4.3.10_, the Framework provides Jackson Serializer and Deserializer implementations for `Message`s and `MessageHeaders` - `MessageHeadersJacksonSerializer` and `MessageJacksonDeserializer`, respectively.
+They have to be configured via the `SimpleModule` options for the `ObjectMapper`.
+In addition, `enableDefaultTyping` should be configured on the `ObjectMapper` to add type information for each serialized complex object.
+That type information is then used during deserialization.
+The Framework provides a utility method `JacksonJsonUtils.messagingAwareMapper()`, which is already supplied with all the above-mentioned properties and serializers.
+To manage JSON serialization in the `RedisMessageStore`, it must be configured like so:
+
+[source,java]
+----
+RedisMessageStore store = new RedisMessageStore(jedisConnectionFactory);
+ObjectMapper mapper = JacksonJsonUtils.messagingAwareMapper();
+RedisSerializer<Object> serializer = new GenericJackson2JsonRedisSerializer(mapper);
+store.setValueSerializer(serializer);
+----
+
 [[redis-cms]]
 ==== Redis Channel Message Stores