From 376405c39b1d96779940f83cdfc061431dd277b8 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Tue, 2 May 2017 13:33:45 -0400 Subject: [PATCH] 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. --- .../integration/message/AdviceMessage.java | 15 ++++ .../store/MessageGroupMetadata.java | 22 ++--- .../AdviceMessageJacksonDeserializer.java | 52 ++++++++++++ .../json/ErrorMessageJacksonDeserializer.java | 52 ++++++++++++ .../GenericMessageJacksonDeserializer.java | 48 +++++++++++ .../support/json/JacksonJsonUtils.java | 54 ++++++++++++- .../json/MessageHeadersJacksonSerializer.java | 59 ++++++++++++++ .../json/MessageJacksonDeserializer.java | 80 +++++++++++++++++++ .../MutableMessageJacksonDeserializer.java | 49 ++++++++++++ .../store/RedisMessageGroupStoreTests.java | 43 +++++++++- src/reference/asciidoc/redis.adoc | 17 +++- 11 files changed, 479 insertions(+), 12 deletions(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/support/json/AdviceMessageJacksonDeserializer.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/support/json/ErrorMessageJacksonDeserializer.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/support/json/GenericMessageJacksonDeserializer.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/support/json/MessageHeadersJacksonSerializer.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/support/json/MessageJacksonDeserializer.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/support/json/MutableMessageJacksonDeserializer.java diff --git a/spring-integration-core/src/main/java/org/springframework/integration/message/AdviceMessage.java b/spring-integration-core/src/main/java/org/springframework/integration/message/AdviceMessage.java index 41e1575ac5..af3409fcf6 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/message/AdviceMessage.java +++ b/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 extends GenericMessage { this.inputMessage = inputMessage; } + /** + * A constructor with the {@link MessageHeaders} instance to use. + *

Note: 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; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupMetadata.java b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupMetadata.java index 42e7932701..2ea8394f3b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupMetadata.java +++ b/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 messageIds = new LinkedList(); - private final List messageIds = new LinkedList(); - - 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 messageIdIterator() { return this.messageIds.iterator(); } @@ -92,6 +92,10 @@ public class MessageGroupMetadata implements Serializable { return null; } + public List getMessageIds() { + return Collections.unmodifiableList(this.messageIds); + } + void complete() { this.complete = true; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/json/AdviceMessageJacksonDeserializer.java b/spring-integration-core/src/main/java/org/springframework/integration/support/json/AdviceMessageJacksonDeserializer.java new file mode 100644 index 0000000000..9d616bb65f --- /dev/null +++ b/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> { + + private static final long serialVersionUID = 1L; + + @SuppressWarnings("unchecked") + public AdviceMessageJacksonDeserializer() { + super((Class>) (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(payload, (MessageHeaders) headers, inputMessage); + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/json/ErrorMessageJacksonDeserializer.java b/spring-integration-core/src/main/java/org/springframework/integration/support/json/ErrorMessageJacksonDeserializer.java new file mode 100644 index 0000000000..86bde95887 --- /dev/null +++ b/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 { + + 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); + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/json/GenericMessageJacksonDeserializer.java b/spring-integration-core/src/main/java/org/springframework/integration/support/json/GenericMessageJacksonDeserializer.java new file mode 100644 index 0000000000..d03e54fffb --- /dev/null +++ b/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> { + + private static final long serialVersionUID = 1L; + + @SuppressWarnings("unchecked") + public GenericMessageJacksonDeserializer() { + super((Class>) (Class) GenericMessage.class); + } + + @Override + protected GenericMessage buildMessage(MutableMessageHeaders headers, Object payload, JsonNode root, + DeserializationContext ctxt) throws IOException { + return new GenericMessage(payload, headers); + } +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/json/JacksonJsonUtils.java b/spring-integration-core/src/main/java/org/springframework/integration/support/json/JacksonJsonUtils.java index e26d040463..8cb72d953c 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/json/JacksonJsonUtils.java +++ b/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."); + } + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/json/MessageHeadersJacksonSerializer.java b/spring-integration-core/src/main/java/org/springframework/integration/support/json/MessageHeadersJacksonSerializer.java new file mode 100644 index 0000000000..6dd36b302a --- /dev/null +++ b/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}. + *

+ * 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 { + + 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(value)); + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/json/MessageJacksonDeserializer.java b/spring-integration-core/src/main/java/org/springframework/integration/support/json/MessageJacksonDeserializer.java new file mode 100644 index 0000000000..76c3bb8722 --- /dev/null +++ b/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> extends StdNodeBasedDeserializer { + + private static final long serialVersionUID = 1L; + + private JavaType payloadType = TypeFactory.defaultInstance().constructType(Object.class); + + private ObjectMapper mapper = new ObjectMapper(); + + protected MessageJacksonDeserializer(Class 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 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; + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/json/MutableMessageJacksonDeserializer.java b/spring-integration-core/src/main/java/org/springframework/integration/support/json/MutableMessageJacksonDeserializer.java new file mode 100644 index 0000000000..c3673932fc --- /dev/null +++ b/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> { + + private static final long serialVersionUID = 1L; + + @SuppressWarnings("unchecked") + public MutableMessageJacksonDeserializer() { + super((Class>) (Class) MutableMessage.class); + } + + @Override + protected MutableMessage buildMessage(MutableMessageHeaders headers, Object payload, JsonNode root, + DeserializationContext ctxt) throws IOException { + return new MutableMessage(payload, headers); + } + +} diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisMessageGroupStoreTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisMessageGroupStoreTests.java index d0f1e711a2..794a38ef43 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisMessageGroupStoreTests.java +++ b/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> 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()); + } + } diff --git a/src/reference/asciidoc/redis.adoc b/src/reference/asciidoc/redis.adoc index 933b17ce8e..4e2b6cd59d 100644 --- a/src/reference/asciidoc/redis.adoc +++ b/src/reference/asciidoc/redis.adoc @@ -319,7 +319,7 @@ Handling these events using an `` 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 serializer = new GenericJackson2JsonRedisSerializer(mapper); +store.setValueSerializer(serializer); +---- + [[redis-cms]] ==== Redis Channel Message Stores