From 0fe6a76453c511f819b28d9789b29f2123e7634e Mon Sep 17 00:00:00 2001 From: Sean Brandt Date: Thu, 22 Dec 2011 17:20:23 -0500 Subject: [PATCH] INT-2334 Fix for MongoDbMessageStore storing MessageHistory This fix ensures that MongoDbMessageStore properly stores MessageHistory that is stored in the 'history' header of MessageHeaders. polished contribution from Sean Brandt, added tests fixed header deserialization logic to ensure that objects are converted to a proper type removed write converter --- .../mongodb/store/MongoDbMessageStore.java | 57 ++++++++++- .../store/MongoDbMessageGroupStoreTests.java | 31 ++++++ .../store/MongoDbMessageStoreTests.java | 97 ++++++++++++++++++- 3 files changed, 179 insertions(+), 6 deletions(-) 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 {