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

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<T extends Message<?>> extends S
 		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);
+		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,

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();
 	}
 

spring-integration-redis/src/test/java/org/springframework/integration/redis/store/redis-aggregator-config.xml

@@ -1,9 +1,15 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <beans xmlns="http://www.springframework.org/schema/beans"
-	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-	xmlns:int="http://www.springframework.org/schema/integration"
-	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">
+	   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">
+
+	<context:annotation-config/>
+
+	<bean class="org.springframework.integration.redis.store.RedisMessageGroupStoreTests"/>
 
 	<int:aggregator input-channel="inputChannel" output-channel="outputChannel" message-store="redisStore"/>
 
@@ -13,10 +19,16 @@
 
 	<bean id="redisStore" class="org.springframework.integration.redis.store.RedisMessageStore">
 		<constructor-arg ref="redisConnectionFactory"/>
+		<property name="valueSerializer">
+			<bean class="org.springframework.data.redis.serializer.GenericJackson2JsonRedisSerializer">
+				<constructor-arg value="#{T (org.springframework.integration.support.json.JacksonJsonUtils).messagingAwareMapper()}"/>
+			</bean>
+		</property>
 	</bean>
 
-	<bean id="redisConnectionFactory" class="org.springframework.data.redis.connection.jedis.JedisConnectionFactory">
-		<property name="port" value="#{T(org.springframework.integration.redis.rules.RedisAvailableRule).REDIS_PORT}"/>
+	<bean id="messageGroupStoreReaper" class="org.springframework.integration.store.MessageGroupStoreReaper">
+		<constructor-arg ref="redisStore"/>
+		<property name="timeout" value="0"/>
 	</bean>
 
 </beans>