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

Merge pull request #281 from olegz/INT-2334

Browse Source 
INT-2334 Fix for MongoDbMessageStore storing MessageHistory
This commit is contained in:
Mark Fisher
2011-12-22 23:25:55 -05:00
parent 953d8a62d2 8ddb6d5c6b
commit 1435dc5794
3 changed files with 176 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

53
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();
}
@@ -333,7 +342,6 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore implements Me
target.put(GROUP_TIMESTAMP_KEY, groupTimestamp);
target.put(GROUP_UPDATE_TIMESTAMP_KEY, lastModified);
}
super.write(message, target);
}
@@ -344,7 +352,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) {
@@ -389,6 +398,28 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore implements Me
}
return null;
}
private Map<String, Object> normalizeHeaders(Map<String, Object> headers) {
Map<String, Object> normalizedHeaders = 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");
try {
Class<?> typeClass = ClassUtils.forName(type.toString(), classLoader);
normalizedHeaders.put(headerName, super.read(typeClass, source));
}
catch (Exception e) {
logger.warn("Header '" + headerName + "' could not be deserialized.", e);
}
}
else {
normalizedHeaders.put(headerName, headerValue);
}
}
return normalizedHeaders;
}
}
@@ -406,6 +437,24 @@ 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();
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);
}
obj.put("components", dbList);
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 {