DelayedMessageWrapper cannot be deserialized when using RedisMessageStore and JSON serialization

Author: NaccOll

Diff for: spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java

@@ -97,6 +97,7 @@
  * @author Artem Bilan
  * @author Gary Russell
  * @author Christian Tzolov
+ * @author Youbin Wu
  *
  * @since 1.0.3
  */
@@ -689,7 +690,7 @@ public static final class DelayedMessageWrapper implements Serializable {
 		@SuppressWarnings("serial")
 		private final Message<?> original;
 
-		DelayedMessageWrapper(Message<?> original, long requestDate) {
+		public DelayedMessageWrapper(Message<?> original, long requestDate) {
 			this.original = original;
 			this.requestDate = requestDate;
 		}

Diff for: spring-integration-core/src/main/java/org/springframework/integration/support/json/DelayedMessageWrapperJacksonDeserializer.java

@@ -0,0 +1,63 @@
+/*
+ * Copyright 2017-2024 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
+ *
+ *      https://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 com.fasterxml.jackson.databind.DeserializationContext;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.deser.std.StdNodeBasedDeserializer;
+
+import org.springframework.integration.handler.DelayHandler;
+import org.springframework.messaging.Message;
+import org.springframework.util.Assert;
+
+import java.io.IOException;
+
+/**
+ * A Jackson {@link StdNodeBasedDeserializer} extension for {@link Message} implementations.
+ *
+ * @author Youbin Wu
+ * @since 6.4
+ */
+public class DelayedMessageWrapperJacksonDeserializer extends StdNodeBasedDeserializer<DelayHandler.DelayedMessageWrapper> {
+
+	private static final long serialVersionUID = 1L;
+
+	private ObjectMapper mapper = new ObjectMapper();
+
+	protected DelayedMessageWrapperJacksonDeserializer() {
+		super(DelayHandler.DelayedMessageWrapper.class);
+	}
+
+	public void setMapper(ObjectMapper mapper) {
+		Assert.notNull(mapper, "'mapper' must not be null");
+		this.mapper = mapper;
+	}
+
+	public ObjectMapper getMapper() {
+		return mapper;
+	}
+
+	@Override
+	public DelayHandler.DelayedMessageWrapper convert(JsonNode root, DeserializationContext ctxt)
+			throws IOException {
+		Long requestDate = this.mapper.readValue(root.get("requestDate").traverse(), Long.class);
+		Message<?> message = this.mapper.readValue(root.get("original").traverse(), Message.class);
+		return new DelayHandler.DelayedMessageWrapper(message, requestDate);
+	}
+
+}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2022 the original author or authors.
+ * Copyright 2002-2024 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.
@@ -36,6 +36,7 @@
 import com.fasterxml.jackson.databind.jsontype.TypeIdResolver;
 import com.fasterxml.jackson.databind.module.SimpleModule;
 
+import org.springframework.integration.handler.DelayHandler;
 import org.springframework.integration.message.AdviceMessage;
 import org.springframework.integration.support.MutableMessage;
 import org.springframework.messaging.support.ErrorMessage;
@@ -46,6 +47,7 @@
  *
  * @author Artem Bilan
  * @author Gary Russell
+ * @author Youbin Wu
  *
  * @since 3.0
  *
@@ -63,7 +65,8 @@ public final class JacksonJsonUtils {
 					"org.springframework.integration.support",
 					"org.springframework.integration.message",
 					"org.springframework.integration.store",
-					"org.springframework.integration.history"
+					"org.springframework.integration.history",
+					"org.springframework.integration.handler"
 			);
 
 	private JacksonJsonUtils() {
@@ -96,13 +99,17 @@ public static ObjectMapper messagingAwareMapper(String... trustedPackages) {
 			MutableMessageJacksonDeserializer mutableMessageDeserializer = new MutableMessageJacksonDeserializer();
 			mutableMessageDeserializer.setMapper(mapper);
 
+			DelayedMessageWrapperJacksonDeserializer delayedMessageWrapperJacksonDeserializer = new DelayedMessageWrapperJacksonDeserializer();
+			delayedMessageWrapperJacksonDeserializer.setMapper(mapper);
+
 			SimpleModule simpleModule = new SimpleModule()
 					.addSerializer(new MessageHeadersJacksonSerializer())
 					.addSerializer(new MimeTypeSerializer())
 					.addDeserializer(GenericMessage.class, genericMessageDeserializer)
 					.addDeserializer(ErrorMessage.class, errorMessageDeserializer)
 					.addDeserializer(AdviceMessage.class, adviceMessageDeserializer)
-					.addDeserializer(MutableMessage.class, mutableMessageDeserializer);
+					.addDeserializer(MutableMessage.class, mutableMessageDeserializer)
+					.addDeserializer(DelayHandler.DelayedMessageWrapper.class, delayedMessageWrapperJacksonDeserializer);
 
 			mapper.registerModule(simpleModule);
 			return mapper;

Diff for: spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisChannelMessageStoreTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2022 the original author or authors.
+ * Copyright 2014-2024 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.
@@ -29,9 +29,11 @@
 import org.springframework.data.redis.serializer.GenericJackson2JsonRedisSerializer;
 import org.springframework.integration.IntegrationMessageHeaderAccessor;
 import org.springframework.integration.channel.NullChannel;
+import org.springframework.integration.handler.DelayHandler;
 import org.springframework.integration.history.MessageHistory;
 import org.springframework.integration.redis.RedisContainerTest;
 import org.springframework.integration.store.MessageGroup;
+import org.springframework.integration.support.DefaultMessageBuilderFactory;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.integration.support.MutableMessageBuilder;
 import org.springframework.integration.support.json.JacksonJsonUtils;
@@ -47,6 +49,7 @@
  * @author Gary Russell
  * @author Artem Bilan
  * @author Artem Vozhdayenko
+ * @author Youbin Wu
  *
  * @since 4.0
  *
@@ -201,4 +204,30 @@ void testJsonSerialization() {
 		assertThat(messages.get(0).getHeaders()).containsKeys(MessageHistory.HEADER_NAME);
 	}
 
+	@Test
+	void testDelayJsonSerialization() {
+		RedisChannelMessageStore store = new RedisChannelMessageStore(RedisContainerTest.connectionFactory());
+		ObjectMapper mapper = JacksonJsonUtils.messagingAwareMapper();
+		GenericJackson2JsonRedisSerializer serializer = new GenericJackson2JsonRedisSerializer(mapper);
+		store.setValueSerializer(serializer);
+
+		Message<?> originMessage = new GenericMessage<>(new Date());
+		Message<?> genericMessage = new DefaultMessageBuilderFactory()
+				.withPayload(new DelayHandler.DelayedMessageWrapper(originMessage, System.currentTimeMillis()))
+				.copyHeaders(originMessage.getHeaders())
+				.build();
+		NullChannel testComponent = new NullChannel();
+		testComponent.setBeanName("testChannel");
+		genericMessage = MessageHistory.write(genericMessage, testComponent);
+
+		String groupId = "delayJsonMessagesStore";
+
+		store.addMessageToGroup(groupId, genericMessage);
+		MessageGroup messageGroup = store.getMessageGroup(groupId);
+		assertThat(messageGroup.size()).isEqualTo(1);
+		List<Message<?>> messages = new ArrayList<>(messageGroup.getMessages());
+		assertThat(messages.get(0)).isEqualTo(genericMessage);
+		assertThat(messages.get(0).getHeaders()).containsKeys(MessageHistory.HEADER_NAME);
+	}
+
 }

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

@@ -42,11 +42,13 @@
 import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.channel.NullChannel;
 import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.handler.DelayHandler;
 import org.springframework.integration.history.MessageHistory;
 import org.springframework.integration.message.AdviceMessage;
 import org.springframework.integration.redis.RedisContainerTest;
 import org.springframework.integration.store.MessageGroup;
 import org.springframework.integration.store.SimpleMessageGroup;
+import org.springframework.integration.support.DefaultMessageBuilderFactory;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.integration.support.MutableMessage;
 import org.springframework.integration.support.json.JacksonJsonUtils;
@@ -64,6 +66,7 @@
  * @author Artem Bilan
  * @author Gary Russell
  * @author Artem Vozhdayenko
+ * @author Youbin Wu
  */
 class RedisMessageGroupStoreTests implements RedisContainerTest {
 
@@ -400,16 +403,21 @@ void testJsonSerialization() {
 		Message<?> mutableMessage = new MutableMessage<>(UUID.randomUUID());
 		Message<?> adviceMessage = new AdviceMessage<>("foo", genericMessage);
 		ErrorMessage errorMessage = new ErrorMessage(new RuntimeException("test exception"), mutableMessage);
+		Message<?> delayMessage = new DefaultMessageBuilderFactory()
+				.withPayload(new DelayHandler.DelayedMessageWrapper(genericMessage, System.currentTimeMillis()))
+				.copyHeaders(genericMessage.getHeaders())
+				.build();
 
-		store.addMessagesToGroup(this.groupId, genericMessage, mutableMessage, adviceMessage, errorMessage);
+		store.addMessagesToGroup(this.groupId, genericMessage, mutableMessage, adviceMessage, errorMessage, delayMessage);
 
 		MessageGroup messageGroup = store.getMessageGroup(this.groupId);
-		assertThat(messageGroup.size()).isEqualTo(4);
+		assertThat(messageGroup.size()).isEqualTo(5);
 		List<Message<?>> messages = new ArrayList<>(messageGroup.getMessages());
 		assertThat(messages.get(0)).isEqualTo(genericMessage);
 		assertThat(messages.get(0).getHeaders()).containsKeys(MessageHistory.HEADER_NAME);
 		assertThat(messages.get(1)).isEqualTo(mutableMessage);
 		assertThat(messages.get(2)).isEqualTo(adviceMessage);
+		assertThat(messages.get(4)).isEqualTo(delayMessage);
 		Message<?> errorMessageResult = messages.get(3);
 		assertThat(errorMessageResult.getHeaders()).isEqualTo(errorMessage.getHeaders());
 		assertThat(errorMessageResult).isInstanceOf(ErrorMessage.class);