INT-3070: Fix Delayer with MongoDbMessageStore

The `DelayHandler.DelayedMessageWrapper` works correctly, as designed by INT-3049.
However `DelayHandler.DelayedMessageWrapper` contains `Message` as a property
and `GenericMessage` can't be reconstructed automatically.

So, added new `DBObjectToGenericMessageConverter` to read `GenericMessage` for
`DelayHandler.DelayedMessageWrapper#original` property.

JIRA: https://jira.springsource.org/browse/INT-3070

Cherry-picked 2.2.x

INT-3070: PR comments

INT-3070: Remove `assert` for common state

spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoDbMessageStore.java

@@ -63,6 +63,7 @@ import com.mongodb.BasicDBList;
 import com.mongodb.BasicDBObject;
 import com.mongodb.DBObject;
 
+
 /**
  * An implementation of both the {@link MessageStore} and {@link MessageGroupStore}
  * strategies that relies upon MongoDB for persistence.
@@ -72,6 +73,7 @@ import com.mongodb.DBObject;
  * @author Sean Brandt
  * @author Jodie StJohn
  * @author Gary Russell
+ * @author Artem Bilan
  * @since 2.1
  */
 public class MongoDbMessageStore extends AbstractMessageGroupStore implements MessageStore, BeanClassLoaderAware {
@@ -313,6 +315,7 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore implements Me
 			customConverters.add(new UuidToDBObjectConverter());
 			customConverters.add(new DBObjectToUUIDConverter());
 			customConverters.add(new MessageHistoryToDBObjectConverter());
+			customConverters.add(new DBObjectToGenericMessageConverter());
 			this.setCustomConversions(new CustomConversions(customConverters));
 			super.afterPropertiesSet();
 		}
@@ -372,7 +375,7 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore implements Me
 				}
 
 				if (completeGroup != null){
-					wrapper.set_Group_complete(completeGroup.booleanValue());
+					wrapper.set_Group_complete(completeGroup);
 				}
 
 				return (S) wrapper;
@@ -414,8 +417,7 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore implements Me
 
 	private static class DBObjectToUUIDConverter implements Converter<DBObject, UUID> {
 		public UUID convert(DBObject source) {
-			UUID id = UUID.fromString((String) source.get("_value"));
-			return id;
+			return UUID.fromString((String) source.get("_value"));
 		}
 	}
 
@@ -438,6 +440,38 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore implements Me
 		}
 	}
 
+	private class DBObjectToGenericMessageConverter implements Converter<DBObject, GenericMessage<?>> {
+
+		@SuppressWarnings("unchecked")
+		public GenericMessage<?> convert(DBObject source) {
+			MessageReadingMongoConverter converter = (MessageReadingMongoConverter) MongoDbMessageStore.this.template
+					.getConverter();
+			Map<String, Object> headers = converter.normalizeHeaders((Map<String, Object>) source.get("headers"));
+
+			Object payload = source.get("payload");
+			Object payloadType = source.get(PAYLOAD_TYPE_KEY);
+			if (payloadType != null && payload instanceof DBObject) {
+				try {
+					Class<?> payloadClass = ClassUtils.forName(payloadType.toString(), classLoader);
+					payload = converter.read(payloadClass, (DBObject) payload);
+				}
+				catch (Exception e) {
+					throw new IllegalStateException("failed to load class: " + payloadType, e);
+				}
+			}
+
+			@SuppressWarnings("rawtypes")
+			GenericMessage<Object> message = new GenericMessage(payload, headers);
+			Map<String, Object> innerMap = (Map<String, Object>) new DirectFieldAccessor(message.getHeaders()).getPropertyValue("headers");
+			// using reflection to set ID and TIMESTAMP since they are immutable through MessageHeaders
+			innerMap.put(MessageHeaders.ID, headers.get(MessageHeaders.ID));
+			innerMap.put(MessageHeaders.TIMESTAMP, headers.get(MessageHeaders.TIMESTAMP));
+
+			return message;
+		}
+
+	}
+
 	/**
 	 * Wrapper class used for storing Messages in MongoDB along with their "group" metadata.
 	 */

spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/DelayerHandlerRescheduleIntegrationTests-context.xml

@@ -0,0 +1,29 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans:beans xmlns:beans="http://www.springframework.org/schema/beans"
+			 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+			 xmlns="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">
+
+    <beans:bean id="mongoConnectionFactory" class="org.springframework.data.mongodb.core.SimpleMongoDbFactory">
+        <beans:constructor-arg>
+            <beans:bean class="com.mongodb.Mongo"/>
+        </beans:constructor-arg>
+        <beans:constructor-arg value="test"/>
+    </beans:bean>
+
+    <beans:bean id="messageStore" class="org.springframework.integration.mongodb.store.MongoDbMessageStore">
+        <beans:constructor-arg ref="mongoConnectionFactory"/>
+    </beans:bean>
+
+    <channel id="output">
+		<queue/>
+	</channel>
+
+	<delayer id="#{T (org.springframework.integration.mongodb.store.DelayerHandlerRescheduleIntegrationTests).DELAYER_ID}"
+			 input-channel="input"
+			 output-channel="output"
+			 default-delay="500"
+			 message-store="messageStore"/>
+
+</beans:beans>

spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/DelayerHandlerRescheduleIntegrationTests.java

@@ -0,0 +1,111 @@
+/*
+ * Copyright 2013 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.mongodb.store;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNotSame;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.util.Iterator;
+
+import org.hamcrest.Matchers;
+import org.junit.Test;
+import org.springframework.context.support.AbstractApplicationContext;
+import org.springframework.context.support.ClassPathXmlApplicationContext;
+import org.springframework.integration.Message;
+import org.springframework.integration.MessageChannel;
+import org.springframework.integration.core.PollableChannel;
+import org.springframework.integration.handler.DelayHandler;
+import org.springframework.integration.mongodb.rules.MongoDbAvailable;
+import org.springframework.integration.mongodb.rules.MongoDbAvailableTests;
+import org.springframework.integration.store.MessageGroup;
+import org.springframework.integration.store.MessageGroupStore;
+import org.springframework.integration.support.MessageBuilder;
+
+/**
+ * @author Artem Bilan
+ * @since 3.0
+ */
+public class DelayerHandlerRescheduleIntegrationTests extends MongoDbAvailableTests {
+
+	public static final String DELAYER_ID = "delayerWithMongoMS";
+
+	@Test
+	@MongoDbAvailable
+	@SuppressWarnings("unchecked")
+	public void testDelayerHandlerRescheduleWithMongoDbMessageStore() throws Exception {
+		AbstractApplicationContext context = new ClassPathXmlApplicationContext(
+				"DelayerHandlerRescheduleIntegrationTests-context.xml", this.getClass());
+		MessageChannel input = context.getBean("input", MessageChannel.class);
+		MessageGroupStore messageStore = context.getBean("messageStore", MessageGroupStore.class);
+
+		String delayerMessageGroupId = DELAYER_ID + ".messageGroupId";
+
+		messageStore.removeMessageGroup(delayerMessageGroupId);
+
+		Message<String> message1 = MessageBuilder.withPayload("test1").build();
+		input.send(message1);
+		input.send(MessageBuilder.withPayload("test2").build());
+
+		// Emulate restart and check DB state before next start
+		context.destroy();
+
+		Thread.sleep(100);
+
+		try {
+			context.getBean("input", MessageChannel.class);
+			fail("IllegalStateException expected");
+		}
+		catch (Exception e) {
+			assertTrue(e instanceof IllegalStateException);
+			assertTrue(e.getMessage().contains("BeanFactory not initialized or already closed - call 'refresh'"));
+		}
+
+		assertEquals(2, messageStore.messageGroupSize(delayerMessageGroupId));
+
+		MessageGroup messageGroup = messageStore.getMessageGroup(delayerMessageGroupId);
+		Iterator<Message<?>> iterator = messageGroup.getMessages().iterator();
+		Message<?> messageInStore = iterator.next();
+		Object payload = messageInStore.getPayload();
+
+		// INT-3049
+		assertTrue(payload instanceof DelayHandler.DelayedMessageWrapper);
+
+		Message<String> original1 = (Message<String>) ((DelayHandler.DelayedMessageWrapper) payload).getOriginal();
+		messageInStore = iterator.next();
+		Message<String> original2 = (Message<String>) ((DelayHandler.DelayedMessageWrapper) messageInStore.getPayload()).getOriginal();
+		assertThat(message1, Matchers.anyOf(Matchers.is(original1), Matchers.is(original2)));
+
+		context.refresh();
+
+		PollableChannel output = context.getBean("output", PollableChannel.class);
+
+		Message<?> message = output.receive(10000);
+		assertNotNull(message);
+
+		Object payload1 = message.getPayload();
+
+		message = output.receive(10000);
+		assertNotNull(message);
+		Object payload2 = message.getPayload();
+		assertNotSame(payload1, payload2);
+
+		assertEquals(0, messageStore.messageGroupSize(delayerMessageGroupId));
+
+	}
+
+}