Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-2334 Fix for MongoDbMessageStore storing MessageHistory

Browse Source 
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
This commit is contained in:
Sean Brandt
2011-12-22 17:20:23 -05:00
committed by Mark Fisher
parent 953d8a62d2
commit 0fe6a76453
3 changed files with 179 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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<Converter<?, ?>> customConverters = new ArrayList<Converter<?,?>>();
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<String, Object> headers = (Map<String, Object>) source.get("headers");
Map<String, Object> headers = this.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) {
@@ -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<String, Object> normalizeHeaders(Map<String, Object> headers){
Map<String, Object> newHeaders= new HashMap<String, Object>();
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<UUID, String> {
@@ -405,6 +439,23 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore implements Me
}
}
private static class MessageHistoryToDBObjectConverter implements Converter<MessageHistory,DBObject> {
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.

31
spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/MongoDbMessageGroupStoreTests.java
View File

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

97
spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/MongoDbMessageStoreTests.java
View File

@@ -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 {