DelayedMessageWrapper cannot be deserialized when using RedisMessageStore and JSON serialization

Author: NaccOll

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

@@ -31,6 +31,8 @@
 import java.util.concurrent.locks.ReentrantLock;
 import java.util.stream.Stream;
 
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
 import org.aopalliance.aop.Advice;
 
 import org.springframework.aop.framework.ProxyFactory;
@@ -97,6 +99,7 @@
  * @author Artem Bilan
  * @author Gary Russell
  * @author Christian Tzolov
+ * @author Youbin Wu
  *
  * @since 1.0.3
  */
@@ -689,7 +692,8 @@ public static final class DelayedMessageWrapper implements Serializable {
 		@SuppressWarnings("serial")
 		private final Message<?> original;
 
-		DelayedMessageWrapper(Message<?> original, long requestDate) {
+		@JsonCreator(mode = JsonCreator.Mode.PROPERTIES)
+		DelayedMessageWrapper(@JsonProperty("original") Message<?> original, @JsonProperty("requestDate") long requestDate) {
 			this.original = original;
 			this.requestDate = requestDate;
 		}

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.
@@ -46,6 +46,7 @@
  *
  * @author Artem Bilan
  * @author Gary Russell
+ * @author Youbin Wu
  *
  * @since 3.0
  *
@@ -63,7 +64,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() {

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

@@ -35,13 +35,15 @@
 import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 
+import org.springframework.beans.BeanUtils;
 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.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;
@@ -64,6 +66,7 @@
  * @author Artem Bilan
  * @author Gary Russell
  * @author Artem Vozhdayenko
+ * @author Youbin Wu
  */
 class RedisMessageGroupStoreTests implements RedisContainerTest {
 
@@ -400,11 +403,13 @@ void testJsonSerialization() {
 		Message<?> mutableMessage = new MutableMessage<>(UUID.randomUUID());
 		Message<?> adviceMessage = new AdviceMessage<>("foo", genericMessage);
 		ErrorMessage errorMessage = new ErrorMessage(new RuntimeException("test exception"), mutableMessage);
+		var delayedMessageWrapperConstructor = BeanUtils.getResolvableConstructor(DelayHandler.DelayedMessageWrapper.class);
+		Message<?> delayMessage = new GenericMessage<>(BeanUtils.instantiateClass(delayedMessageWrapperConstructor, genericMessage, System.currentTimeMillis()));
 
-		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);
@@ -417,6 +422,7 @@ void testJsonSerialization() {
 				.isEqualTo(errorMessage.getOriginalMessage());
 		assertThat(((ErrorMessage) errorMessageResult).getPayload().getMessage())
 				.isEqualTo(errorMessage.getPayload().getMessage());
+		assertThat(messages.get(4)).isEqualTo(delayMessage);
 
 		Message<Foo> fooMessage = new GenericMessage<>(new Foo("foo"));
 		try {