From c877b041d4d2cb3176f640285199580a37ed99a9 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Thu, 17 Aug 2017 14:40:32 -0400 Subject: [PATCH] GH-2217: JSON: Populated actual ID & TIMESTAMP Fixes spring-projects/spring-integration#2217 The `MutableMessageHeaders` delegates provided headers to the super which just generates fresh `ID` & `TIMESTAMP` headers ignoring provided. * Add explicit `ID` & `TIMESTAMP` headers population in the `MessageJacksonDeserializer` after creation `MutableMessageHeaders`. We can't apply the fix for the `MutableMessageHeaders` like it is in the `master` because it would be a breaking change --- .../json/MessageJacksonDeserializer.java | 6 ++++- .../store/RedisMessageGroupStoreTests.java | 19 +++++++++++++-- .../redis/store/redis-aggregator-config.xml | 24 ++++++++++++++----- 3 files changed, 40 insertions(+), 9 deletions(-) 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 index 76c3bb8722..33d3d8da72 100644 --- 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 @@ -22,6 +22,7 @@ import java.util.Map; import org.springframework.integration.support.MutableMessageHeaders; import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHeaders; import org.springframework.util.Assert; import com.fasterxml.jackson.annotation.JsonTypeInfo; @@ -71,7 +72,10 @@ public abstract class MessageJacksonDeserializer> extends S 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); + MutableMessageHeaders messageHeaders = new MutableMessageHeaders(headers); + messageHeaders.put(MessageHeaders.ID, headers.get(MessageHeaders.ID)); + messageHeaders.put(MessageHeaders.TIMESTAMP, headers.get(MessageHeaders.TIMESTAMP)); + return buildMessage(messageHeaders, payload, root, ctxt); } protected abstract T buildMessage(MutableMessageHeaders headers, Object payload, JsonNode root, 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 df4a8c7fa0..a84d7354c3 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 @@ -42,6 +42,7 @@ import org.junit.Before; import org.junit.Ignore; import org.junit.Test; +import org.springframework.context.annotation.Bean; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.data.redis.connection.RedisConnectionFactory; import org.springframework.data.redis.core.StringRedisTemplate; @@ -53,6 +54,8 @@ 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.MessageGroupStore; +import org.springframework.integration.store.MessageGroupStoreReaper; import org.springframework.integration.store.SimpleMessageGroup; import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.support.MutableMessage; @@ -73,6 +76,11 @@ import junit.framework.AssertionFailedError; */ public class RedisMessageGroupStoreTests extends RedisAvailableTests { + @Bean + public RedisConnectionFactory redisConnectionFactory() { + return getConnectionFactoryForTest(); + } + @Before @After public void setUpTearDown() { @@ -393,9 +401,9 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests { .build(); input.send(m1); - assertNull(output.receive(1000)); + assertNull(output.receive(10)); input.send(m2); - assertNull(output.receive(1000)); + assertNull(output.receive(10)); context.close(); @@ -411,6 +419,13 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests { input.send(m3); assertNotNull(output.receive(1000)); + + MessageGroupStoreReaper messageGroupStoreReaper = context.getBean(MessageGroupStoreReaper.class); + messageGroupStoreReaper.run(); + + MessageGroupStore messageGroupStore = context.getBean(MessageGroupStore.class); + assertEquals(0, messageGroupStore.getMessageGroupCount()); + context.close(); } diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/redis-aggregator-config.xml b/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/redis-aggregator-config.xml index 612e43ef4e..e8b28d346b 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/redis-aggregator-config.xml +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/redis-aggregator-config.xml @@ -1,9 +1,15 @@ + xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" + xmlns:int="http://www.springframework.org/schema/integration" + xmlns:context="http://www.springframework.org/schema/context" + xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd + http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd + http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd"> + + + + @@ -13,10 +19,16 @@ + + + + + - - + + +