diff --git a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoDbMessageStore.java b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoDbMessageStore.java index 274e3fbf95..5546d1004a 100644 --- a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoDbMessageStore.java +++ b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoDbMessageStore.java @@ -21,6 +21,7 @@ import java.util.HashMap; import java.util.Iterator; import java.util.List; import java.util.Map; +import java.util.Properties; import java.util.UUID; import org.springframework.beans.DirectFieldAccessor; @@ -39,6 +40,7 @@ import org.springframework.data.mongodb.core.query.Query; import org.springframework.data.mongodb.core.query.Update; import org.springframework.integration.Message; import org.springframework.integration.MessageHeaders; +import org.springframework.integration.history.MessageHistory; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.store.AbstractMessageGroupStore; import org.springframework.integration.store.MessageGroup; @@ -51,9 +53,14 @@ import org.springframework.util.ClassUtils; import org.springframework.util.CollectionUtils; import org.springframework.util.StringUtils; +import com.mongodb.BasicDBList; +import com.mongodb.BasicDBObject; import com.mongodb.DBObject; import static org.springframework.data.mongodb.core.query.Criteria.where; +import static org.springframework.integration.history.MessageHistory.NAME_PROPERTY; +import static org.springframework.integration.history.MessageHistory.TIMESTAMP_PROPERTY; +import static org.springframework.integration.history.MessageHistory.TYPE_PROPERTY; /** * An implementation of both the {@link MessageStore} and {@link MessageGroupStore} @@ -61,6 +68,7 @@ import static org.springframework.data.mongodb.core.query.Criteria.where; * * @author Mark Fisher * @author Oleg Zhurakousky + * @author Sean Brandt * @since 2.1 */ public class MongoDbMessageStore extends AbstractMessageGroupStore implements MessageStore, BeanClassLoaderAware { @@ -298,6 +306,7 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore implements Me List> customConverters = new ArrayList>(); customConverters.add(new UuidToStringConverter()); customConverters.add(new StringToUuidConverter()); + customConverters.add(new MessageHistoryToDBObjectConverter()); this.setCustomConversions(new CustomConversions(customConverters)); super.afterPropertiesSet(); } @@ -344,7 +353,8 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore implements Me return super.read(clazz, source); } if (source != null) { - Map headers = (Map) source.get("headers"); + Map headers = this.normalizeHeaders((Map) source.get("headers")); + Object payload = source.get("payload"); Object payloadType = source.get(PAYLOAD_TYPE_KEY); if (payloadType != null && payload instanceof DBObject) { @@ -356,7 +366,7 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore implements Me throw new IllegalStateException("failed to load class: " + payloadType, e); } } - GenericMessage message = new GenericMessage(payload, headers); + GenericMessage message = new GenericMessage(payload, headers); Map innerMap = (Map) new DirectFieldAccessor(message.getHeaders()).getPropertyValue("headers"); // using reflection to set ID and TIMESTAMP since they are immutable through MessageHeaders innerMap.put(MessageHeaders.ID, UUID.fromString((String) headers.get(MessageHeaders.ID))); @@ -389,7 +399,31 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore implements Me } return null; } - } + + private Map normalizeHeaders(Map headers){ + Map newHeaders= new HashMap(); + for (String headerName : headers.keySet()) { + Object headerValue = headers.get(headerName); + if (headerValue instanceof DBObject){ + DBObject source = (DBObject) headerValue; + Object type = source.get("_class"); + if (type != null){ + try { + Class typeClass = ClassUtils.forName(type.toString(), classLoader); + Object obj = super.read(typeClass, source); + newHeaders.put(headerName, obj); + } catch (Exception e) { + logger.warn("Header '" + headerName + "' could not be deserialized due to exception: ", e); + } + } + } + else { + newHeaders.put(headerName, headerValue); + } + } + return newHeaders; + } + } private static class UuidToStringConverter implements Converter { @@ -405,6 +439,23 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore implements Me } } + private static class MessageHistoryToDBObjectConverter implements Converter { + + public DBObject convert(MessageHistory source) { + BasicDBObject obj = new BasicDBObject(); + obj.put("_class", MessageHistory.class.getName()); + BasicDBList dbList = new BasicDBList(); + obj.put("components", dbList); + for (Properties properties : source) { + BasicDBObject dbo = new BasicDBObject(); + dbo.put(NAME_PROPERTY, properties.getProperty(NAME_PROPERTY)); + dbo.put(TYPE_PROPERTY, properties.getProperty(TYPE_PROPERTY)); + dbo.put(TIMESTAMP_PROPERTY, properties.getProperty(TIMESTAMP_PROPERTY)); + dbList.add(dbo); + } + return obj; + } + } /** * Wrapper class used for storing Messages in MongoDB along with their "group" metadata. diff --git a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/MongoDbMessageGroupStoreTests.java b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/MongoDbMessageGroupStoreTests.java index cb81f41be3..e3d07a56af 100644 --- a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/MongoDbMessageGroupStoreTests.java +++ b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/MongoDbMessageGroupStoreTests.java @@ -16,6 +16,7 @@ package org.springframework.integration.mongodb.store; import java.util.Iterator; +import java.util.Properties; import java.util.UUID; import org.junit.Test; @@ -24,7 +25,9 @@ import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.data.mongodb.MongoDbFactory; import org.springframework.integration.Message; import org.springframework.integration.MessageChannel; +import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.history.MessageHistory; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.mongodb.rules.MongoDbAvailable; import org.springframework.integration.mongodb.rules.MongoDbAvailableTests; @@ -332,4 +335,32 @@ public class MongoDbMessageGroupStoreTests extends MongoDbAvailableTests { assertNotNull(output.receive(2000)); } + @Test + @MongoDbAvailable + public void testWithMessageHistory() throws Exception{ + MongoDbFactory mongoDbFactory = this.prepareMongoFactory(); + MongoDbMessageStore store = new MongoDbMessageStore(mongoDbFactory); + + store.getMessageGroup(1); + + Message message = new GenericMessage("Hello"); + DirectChannel fooChannel = new DirectChannel(); + fooChannel.setBeanName("fooChannel"); + DirectChannel barChannel = new DirectChannel(); + barChannel.setBeanName("barChannel"); + + message = MessageHistory.write(message, fooChannel); + message = MessageHistory.write(message, barChannel); + store.addMessageToGroup(1, message); + + message = store.getMessageGroup(1).getMessages().iterator().next(); + + MessageHistory messageHistory = MessageHistory.read(message); + assertNotNull(messageHistory); + assertEquals(2, messageHistory.size()); + Properties fooChannelHistory = messageHistory.get(0); + assertEquals("fooChannel", fooChannelHistory.get("name")); + assertEquals("channel", fooChannelHistory.get("type")); + } + } diff --git a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/MongoDbMessageStoreTests.java b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/MongoDbMessageStoreTests.java index 89b6970522..19c444f87c 100644 --- a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/MongoDbMessageStoreTests.java +++ b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/MongoDbMessageStoreTests.java @@ -16,21 +16,26 @@ package org.springframework.integration.mongodb.store; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertNotNull; -import static org.junit.Assert.assertNull; +import java.util.Properties; import org.junit.Test; import org.springframework.data.mongodb.MongoDbFactory; import org.springframework.data.mongodb.core.SimpleMongoDbFactory; import org.springframework.integration.Message; +import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.history.MessageHistory; import org.springframework.integration.mongodb.rules.MongoDbAvailable; import org.springframework.integration.mongodb.rules.MongoDbAvailableTests; import org.springframework.integration.support.MessageBuilder; import com.mongodb.Mongo; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; + /** * @author Mark Fisher * @author Oleg Zhurakousky @@ -80,6 +85,92 @@ public class MongoDbMessageStoreTests extends MongoDbAvailableTests{ assertEquals(messageToStore.getHeaders(), retrievedMessage.getHeaders()); assertEquals(messageToStore, retrievedMessage); } + + @Test + @MongoDbAvailable + public void testWithMessageHistory() throws Exception{ + MongoDbFactory mongoDbFactory = this.prepareMongoFactory(); + MongoDbMessageStore store = new MongoDbMessageStore(mongoDbFactory); + + Foo foo = new Foo(); + foo.setName("foo"); + Message message = MessageBuilder.withPayload(foo). + setHeader("foo", foo). + setHeader("bar", new Bar("bar")). + setHeader("baz", new Baz()). + setHeader("abc", new Abc()). + setHeader("xyz", new Xyz()). + build(); + DirectChannel fooChannel = new DirectChannel(); + fooChannel.setBeanName("fooChannel"); + DirectChannel barChannel = new DirectChannel(); + barChannel.setBeanName("barChannel"); + + message = MessageHistory.write(message, fooChannel); + message = MessageHistory.write(message, barChannel); + store.addMessage(message); + message = store.getMessage(message.getHeaders().getId()); + assertTrue(message.getHeaders().get("foo") instanceof Foo); + assertTrue(message.getHeaders().get("bar") instanceof Bar); + assertTrue(message.getHeaders().get("baz") instanceof Baz); + assertTrue(message.getHeaders().get("abc") instanceof Abc); + assertTrue(message.getHeaders().get("xyz") instanceof Xyz); + MessageHistory messageHistory = MessageHistory.read(message); + assertNotNull(messageHistory); + assertEquals(2, messageHistory.size()); + Properties fooChannelHistory = messageHistory.get(0); + assertEquals("fooChannel", fooChannelHistory.get("name")); + assertEquals("channel", fooChannelHistory.get("type")); + } + + public static class Foo{ + private String name; + + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } + } + + public static class Bar{ + private String name; + + public Bar(String name){ + this.name = name; + } + + public String getName() { + return name; + } + } + + public static class Baz{ + private String name = "baz"; + + public String getName() { + return name; + } + } + + public static class Abc{ + private String name = "abx"; + + private Abc(){} + + public String getName() { + return name; + } + } + + public static class Xyz{ + @SuppressWarnings("unused") + private String name = "xyz"; + + private Xyz(){} + } public static class Person {