From d2bb90f7e0c0a98674b3021618030bda30e3c6d6 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Thu, 14 Nov 2013 18:57:47 +0200 Subject: [PATCH] INT-3143: Add `ConfigurableMongoDbMessageStore` JIRA: https://jira.springsource.org/browse/INT-3143 INT-3143: Polishing, fixed and documentation JIRA: https://jira.springsource.org/browse/INT-3077 --- .../ConfigurableMongoDbMessageStore.java | 456 ++++++++++++++++ .../mongodb/rules/MongoDbAvailableTests.java | 41 +- ...AbstractMongoDbMessageGroupStoreTests.java | 505 ++++++++++++++++++ .../AbstractMongoDbMessageStoreTests.java | 295 ++++++++++ ...igurableMongoDbMessageGroupStoreTests.java | 103 ++++ ...bleMongoDbMessageStore-CustomConverter.xml | 45 ++ .../ConfigurableMongoDbMessageStoreTests.java | 123 +++++ ...leIntegrationConfigurableTests-context.xml | 29 + ...ayerHandlerRescheduleIntegrationTests.java | 18 +- .../store/MongoDbMessageGroupStoreTests.java | 465 +--------------- ...essageStoreClaimCheckIntegrationTests.java | 68 ++- .../store/MongoDbMessageStoreTests.java | 235 +------- .../mongo-aggregator-confugurable-config.xml | 25 + src/reference/docbook/mongodb.xml | 29 +- src/reference/docbook/whats-new.xml | 9 + 15 files changed, 1739 insertions(+), 707 deletions(-) create mode 100644 spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/ConfigurableMongoDbMessageStore.java create mode 100644 spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/AbstractMongoDbMessageGroupStoreTests.java create mode 100644 spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/AbstractMongoDbMessageStoreTests.java create mode 100644 spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/ConfigurableMongoDbMessageGroupStoreTests.java create mode 100644 spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/ConfigurableMongoDbMessageStore-CustomConverter.xml create mode 100644 spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/ConfigurableMongoDbMessageStoreTests.java create mode 100644 spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/DelayerHandlerRescheduleIntegrationConfigurableTests-context.xml create mode 100644 spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/mongo-aggregator-confugurable-config.xml diff --git a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/ConfigurableMongoDbMessageStore.java b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/ConfigurableMongoDbMessageStore.java new file mode 100644 index 0000000000..fecae0c12c --- /dev/null +++ b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/ConfigurableMongoDbMessageStore.java @@ -0,0 +1,456 @@ +/* + * 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 java.util.ArrayList; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.UUID; + +import org.springframework.beans.BeansException; +import org.springframework.beans.DirectFieldAccessor; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.context.ApplicationContext; +import org.springframework.context.ApplicationContextAware; +import org.springframework.core.convert.TypeDescriptor; +import org.springframework.core.convert.converter.Converter; +import org.springframework.core.convert.converter.GenericConverter; +import org.springframework.core.serializer.support.DeserializingConverter; +import org.springframework.core.serializer.support.SerializingConverter; +import org.springframework.data.domain.Sort; +import org.springframework.data.mongodb.MongoDbFactory; +import org.springframework.data.mongodb.core.IndexOperations; +import org.springframework.data.mongodb.core.MongoTemplate; +import org.springframework.data.mongodb.core.convert.CustomConversions; +import org.springframework.data.mongodb.core.convert.MappingMongoConverter; +import org.springframework.data.mongodb.core.index.Index; +import org.springframework.data.mongodb.core.mapping.MongoMappingContext; +import org.springframework.data.mongodb.core.query.Criteria; +import org.springframework.data.mongodb.core.query.Order; +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.store.AbstractMessageGroupStore; +import org.springframework.integration.store.MessageGroup; +import org.springframework.integration.store.MessageGroupStore; +import org.springframework.integration.store.MessageStore; +import org.springframework.integration.store.SimpleMessageGroup; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.util.Assert; + +/** + * An alternate MongoDB {@link MessageStore} and {@link MessageGroupStore} which allows the user to + * configure the instance of {@link MongoTemplate}. The mechanism of storing the messages/group of messages + * in the store is and is different from {@link MongoDbMessageStore}. Since the store uses serialization of the + * messages by default, all the headers, and the payload of the Message must implement {@link java.io.Serializable} + * interface + * + * @author Amol Nayak + * @author Artem Bilan + * @since 3.0 + */ +public class ConfigurableMongoDbMessageStore extends AbstractMessageGroupStore + implements MessageStore, InitializingBean, ApplicationContextAware { + + public final static String DEFAULT_COLLECTION_NAME = "configurableStoreMessages"; + + /** + * The name of the message header that stores a flag to indicate that the message has been saved. This is an + * optimization for the put method. + */ + public static final String SAVED_KEY = ConfigurableMongoDbMessageStore.class.getSimpleName() + ".SAVED"; + + /** + * The name of the message header that stores a timestamp for the time the message was inserted. + */ + public static final String CREATED_DATE_KEY = ConfigurableMongoDbMessageStore.class.getSimpleName() + ".CREATED_DATE"; + + private static final String MESSAGE_ID = "messageId"; + + private static final String GROUP_ID = "groupId"; + + private static final String LAST_MODIFIED_TIME = "lastModifiedTime"; + + private static final String LAST_RELEASED_SEQUENCE = "lastReleasedSequence"; + + private static final String COMPLETE = "complete"; + + private final String collectionName; + + private final MongoDbFactory mongoDbFactory; + + private volatile MongoTemplate mongoTemplate; + + private volatile MappingMongoConverter mappingMongoConverter; + + private ApplicationContext applicationContext; + + + public ConfigurableMongoDbMessageStore(MongoTemplate mongoTemplate) { + this(mongoTemplate, DEFAULT_COLLECTION_NAME); + } + + public ConfigurableMongoDbMessageStore(MongoTemplate mongoTemplate, String collectionName) { + Assert.notNull("'mongoTemplate' must not be null"); + Assert.hasText("'collectionName' must not be empty"); + this.collectionName = collectionName; + this.mongoTemplate = mongoTemplate; + this.mongoDbFactory = null; + } + + public ConfigurableMongoDbMessageStore(MongoDbFactory mongoDbFactory) { + this(mongoDbFactory, null, DEFAULT_COLLECTION_NAME); + } + + public ConfigurableMongoDbMessageStore(MongoDbFactory mongoDbFactory, MappingMongoConverter mappingMongoConverter) { + this(mongoDbFactory, mappingMongoConverter, DEFAULT_COLLECTION_NAME); + } + + public ConfigurableMongoDbMessageStore(MongoDbFactory mongoDbFactory, String collectionName) { + this(mongoDbFactory, null, collectionName); + } + + public ConfigurableMongoDbMessageStore(MongoDbFactory mongoDbFactory, MappingMongoConverter mappingMongoConverter, String collectionName) { + Assert.notNull("'mongoDbFactory' must not be null"); + Assert.hasText("'collectionName' must not be empty"); + this.collectionName = collectionName; + this.mongoDbFactory = mongoDbFactory; + this.mappingMongoConverter = mappingMongoConverter; + } + + @Override + public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { + this.applicationContext = applicationContext; + } + + @Override + public void afterPropertiesSet() throws Exception { + if (this.mongoTemplate == null) { + if (this.mappingMongoConverter == null) { + this.mappingMongoConverter = new MappingMongoConverter(this.mongoDbFactory, new MongoMappingContext()); + this.mappingMongoConverter.setApplicationContext(this.applicationContext); + List customConverters = new ArrayList(); + customConverters.add(new MongoDbMessageBytesConverter()); + this.mappingMongoConverter.setCustomConversions(new CustomConversions(customConverters)); + this.mappingMongoConverter.afterPropertiesSet(); + } + this.mongoTemplate = new MongoTemplate(this.mongoDbFactory, this.mappingMongoConverter); + if (this.applicationContext != null) { + this.mongoTemplate.setApplicationContext(this.applicationContext); + } + } + + IndexOperations indexOperations = this.mongoTemplate.indexOps(this.collectionName); + indexOperations.ensureIndex(new Index(MESSAGE_ID, Order.ASCENDING)); + indexOperations.ensureIndex(new Index(GROUP_ID, Order.ASCENDING).on(LAST_MODIFIED_TIME, Order.DESCENDING)); + } + + + @Override + public Message getMessage(UUID id) { + Assert.notNull(id, "'id' must not be null"); + MessageDocument document = this.mongoTemplate.findOne(Query.query(Criteria.where(MESSAGE_ID).is(id)), + MessageDocument.class, this.collectionName); + return (document != null) ? document.getMessage() : null; + } + + @Override + public Message addMessage(Message message) { + Assert.notNull(message, "'message' must not be null"); + this.addMessageDocument(new MessageDocument(message)); + return message; + } + + @SuppressWarnings({ "rawtypes", "unchecked" }) + private void addMessageDocument(MessageDocument document) { + Message message = document.getMessage(); + if (message.getHeaders().containsKey(SAVED_KEY)) { + Message saved = getMessage(message.getHeaders().getId()); + if (saved != null) { + if (saved.equals(message)) { + return; + } // We need to save it under its own id + } + } + + final long createdDate = document.getCreatedTime() == 0 ? System.currentTimeMillis() : document.getCreatedTime(); + + Message result = MessageBuilder.fromMessage(message).setHeader(SAVED_KEY, Boolean.TRUE) + .setHeader(CREATED_DATE_KEY, createdDate).build(); + + Map innerMap = (Map) new DirectFieldAccessor(result.getHeaders()).getPropertyValue("headers"); + // using reflection to set ID since it is immutable through MessageHeaders + innerMap.put(MessageHeaders.ID, message.getHeaders().get(MessageHeaders.ID)); + innerMap.put(MessageHeaders.TIMESTAMP, message.getHeaders().get(MessageHeaders.TIMESTAMP)); + + document.setCreatedTime(createdDate); + this.mongoTemplate.insert(document, this.collectionName); + } + + @Override + public Message removeMessage(UUID id) { + Assert.notNull(id, "'id' must not be null"); + MessageDocument document = this.mongoTemplate.findAndRemove(Query.query(Criteria.where(MESSAGE_ID).is(id)), + MessageDocument.class, this.collectionName); + return (document != null) ? document.getMessage() : null; + } + + @Override + public long getMessageCount() { + return this.mongoTemplate.getCollection(this.collectionName).getCount(); + } + + + @Override + public int messageGroupSize(Object groupId) { + long lCount = this.mongoTemplate.count(groupIdQuery(groupId), this.collectionName); + Assert.isTrue(lCount <= Integer.MAX_VALUE, "Message count is out of Integer's range"); + return (int) lCount; + } + + @Override + public MessageGroup getMessageGroup(Object groupId) { + List messageDocuments = this.mongoTemplate.find(groupIdQuery(groupId), MessageDocument.class, + this.collectionName); + + long createdTime = 0; + long lastModifiedTime = 0; + int lastReleasedSequence = 0; + boolean complete = false; + + if (messageDocuments.size() > 0) { + MessageDocument document = messageDocuments.get(0); + createdTime = document.getCreatedTime(); + lastModifiedTime = document.getLastModifiedTime(); + complete = document.isComplete(); + lastReleasedSequence = document.getLastReleasedSequence(); + } + + List> messages = new ArrayList>(); + for (MessageDocument document : messageDocuments) { + messages.add(document.getMessage()); + } + SimpleMessageGroup group = new SimpleMessageGroup(messages, groupId, createdTime, complete); + group.setLastReleasedMessageSequenceNumber(lastReleasedSequence); + group.setLastModified(lastModifiedTime); + + return group; + } + + @Override + public MessageGroup addMessageToGroup(Object groupId, Message message) { + Assert.notNull(groupId, "'groupId' must not be null"); + Assert.notNull(message, "'message' must not be null"); + MessageDocument messageDocument = this.mongoTemplate.findOne(groupIdQuery(groupId), MessageDocument.class, + this.collectionName); + + long createdTime = 0; + int lastReleasedSequence = 0; + boolean complete = false; + + if (messageDocument != null) { + createdTime = messageDocument.getCreatedTime(); + lastReleasedSequence = messageDocument.getLastReleasedSequence(); + complete = messageDocument.isComplete(); + } + + MessageDocument document = new MessageDocument(message); + document.setGroupId(groupId); + document.setComplete(complete); + document.setLastReleasedSequence(lastReleasedSequence); + document.setCreatedTime(createdTime == 0 ? System.currentTimeMillis() : createdTime); + document.setLastModifiedTime(System.currentTimeMillis()); + + this.addMessageDocument(document); + + return this.getMessageGroup(groupId); + } + + @Override + public MessageGroup removeMessageFromGroup(Object groupId, Message messageToRemove) { + Assert.notNull(groupId, "'groupId' must not be null"); + Assert.notNull(messageToRemove, "'messageToRemove' must not be null"); + Query query = groupIdQuery(groupId).addCriteria(Criteria.where(MESSAGE_ID).is(messageToRemove.getHeaders().getId())); + this.mongoTemplate.remove(query, this.collectionName); + this.updateGroup(groupId, lastModifiedUpdate()); + return this.getMessageGroup(groupId); + } + + @Override + public void removeMessageGroup(Object groupId) { + this.mongoTemplate.remove(groupIdQuery(groupId), this.collectionName); + } + + @Override + @SuppressWarnings({ "rawtypes" }) + public Iterator iterator() { + Map messageGroupMap = new HashMap(); + Query query = Query.query(Criteria.where(GROUP_ID).exists(true)); + query.fields().include(GROUP_ID); + List groupIds = this.mongoTemplate.find(query, Map.class, this.collectionName); + for (Map groupId : groupIds) { + Object key = groupId.get(GROUP_ID); + if (!messageGroupMap.containsKey(key)) { + messageGroupMap.put(key, this.getMessageGroup(groupId)); + } + } + return messageGroupMap.values().iterator(); + } + + @Override + public Message pollMessageFromGroup(Object groupId) { + Assert.notNull(groupId, "'groupId' must not be null"); + Query query = groupIdQuery(groupId).with(new Sort(Sort.Direction.ASC, LAST_MODIFIED_TIME)); + MessageDocument document = this.mongoTemplate.findAndRemove(query, MessageDocument.class, this.collectionName); + Message message = null; + if (document != null) { + message = document.getMessage(); + this.updateGroup(groupId, lastModifiedUpdate()); + } + return message; + } + + @Override + public void setLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber) { + this.updateGroup(groupId, lastModifiedUpdate().set(LAST_RELEASED_SEQUENCE, sequenceNumber)); + } + + @Override + public void completeGroup(Object groupId) { + this.updateGroup(groupId, lastModifiedUpdate().set(COMPLETE, true)); + } + + + private void updateGroup(Object groupId, Update update) { + this.mongoTemplate.updateFirst(groupIdQuery(groupId), update, this.collectionName); + } + + private static Update lastModifiedUpdate() { + return Update.update(LAST_MODIFIED_TIME, System.currentTimeMillis()); + } + + + private static Query groupIdQuery(Object groupId) { + return Query.query(Criteria.where(GROUP_ID).is(groupId)); + } + + /** + * The entity class to wrap {@link Message} to the MongoDB document. + */ + private static class MessageDocument { + + private final Message message; + + @SuppressWarnings("unused") + private final UUID messageId; + + private volatile Long createdTime = 0L; + + @SuppressWarnings("unused") + private volatile Object groupId; + + private volatile Long lastModifiedTime = 0L; + + private volatile Boolean complete = false; + + private volatile Integer lastReleasedSequence = 0; + + public MessageDocument(Message message) { + Assert.notNull(message, "'message' must not be null"); + this.message = message; + this.messageId = message.getHeaders().getId(); + } + + public Message getMessage() { + return message; + } + + public void setGroupId(Object groupId) { + this.groupId = groupId; + } + + public Long getLastModifiedTime() { + return lastModifiedTime; + } + + public void setLastModifiedTime(long lastModifiedTime) { + this.lastModifiedTime = lastModifiedTime; + } + + public Long getCreatedTime() { + return createdTime; + } + + public void setCreatedTime(long createdTime) { + this.createdTime = createdTime; + } + + public Boolean isComplete() { + return complete; + } + + public void setComplete(boolean complete) { + this.complete = complete; + } + + public Integer getLastReleasedSequence() { + return lastReleasedSequence; + } + + public void setLastReleasedSequence(int lastReleasedSequence) { + this.lastReleasedSequence = lastReleasedSequence; + } + + } + + /** + * A {@link GenericConverter} implementation to convert {@link Message} to + * serialized {@link byte[]} to store {@link Message} to the MongoDB. + * And vice versa - to convert {@link byte[]} from the MongoDB to the {@link Message}. + */ + private static class MongoDbMessageBytesConverter implements GenericConverter { + + private final Converter serializingConverter = new SerializingConverter(); + + private final Converter deserializingConverter = new DeserializingConverter(); + + @Override + public Set getConvertibleTypes() { + Set convertiblePairs = new HashSet(); + convertiblePairs.add(new ConvertiblePair(Message.class, byte[].class)); + convertiblePairs.add(new ConvertiblePair(byte[].class, Message.class)); + return convertiblePairs; + } + + @Override + public Object convert(Object source, TypeDescriptor sourceType, TypeDescriptor targetType) { + if (Message.class.isAssignableFrom(sourceType.getObjectType())) { + return serializingConverter.convert(source); + } + else { + return deserializingConverter.convert((byte[]) source); + } + } + + } + +} + diff --git a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/rules/MongoDbAvailableTests.java b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/rules/MongoDbAvailableTests.java index 3b2bdbb9ee..28e1ae15d9 100644 --- a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/rules/MongoDbAvailableTests.java +++ b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/rules/MongoDbAvailableTests.java @@ -41,18 +41,23 @@ public abstract class MongoDbAvailableTests { public MongoDbAvailableRule redisAvailableRule = new MongoDbAvailableRule(); - protected MongoDbFactory prepareMongoFactory(String... additionalCollectionToDrop) throws Exception{ + protected MongoDbFactory prepareMongoFactory(String... additionalCollectionsToDrop) throws Exception { MongoDbFactory mongoDbFactory = new SimpleMongoDbFactory(new Mongo(), "test"); - MongoTemplate template = new MongoTemplate(mongoDbFactory); - template.dropCollection("messages"); - template.dropCollection("data"); - for (String additionalCollection : additionalCollectionToDrop) { - template.dropCollection(additionalCollection); - } + cleanupCollections(mongoDbFactory, additionalCollectionsToDrop); return mongoDbFactory; } - public Person createPerson(){ + protected void cleanupCollections(MongoDbFactory mongoDbFactory, String... additionalCollectionsToDrop) { + MongoTemplate template = new MongoTemplate(mongoDbFactory); + template.dropCollection("messages"); + template.dropCollection("configurableStoreMessages"); + template.dropCollection("data"); + for (String additionalCollection : additionalCollectionsToDrop) { + template.dropCollection(additionalCollection); + } + } + + public Person createPerson() { Address address = new Address(); address.setCity("Philadelphia"); address.setStreet("2121 Rawn street"); @@ -64,7 +69,7 @@ public abstract class MongoDbAvailableTests { return person; } - public Person createPerson(String name){ + public Person createPerson(String name) { Address address = new Address(); address.setCity("Philadelphia"); address.setStreet("2121 Rawn street"); @@ -77,44 +82,61 @@ public abstract class MongoDbAvailableTests { } public static class Person { + private Address address; + private String name; + public Address getAddress() { return address; } + public void setAddress(Address address) { this.address = address; } + public String getName() { return name; } + public void setName(String name) { this.name = name; } + } public static class Address { + private String street; + private String city; + private String state; + public String getStreet() { return street; } + public void setStreet(String street) { this.street = street; } + public String getCity() { return city; } + public void setCity(String city) { this.city = city; } + public String getState() { return state; } + public void setState(String state) { this.state = state; } + } public static class TestMongoConverter extends MappingMongoConverter { @@ -134,6 +156,7 @@ public abstract class MongoDbAvailableTests { public S read(Class clazz, DBObject source) { return super.read(clazz, source); } + } } diff --git a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/AbstractMongoDbMessageGroupStoreTests.java b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/AbstractMongoDbMessageGroupStoreTests.java new file mode 100644 index 0000000000..b30c09b05a --- /dev/null +++ b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/AbstractMongoDbMessageGroupStoreTests.java @@ -0,0 +1,505 @@ +/* + * Copyright 2002-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.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; + +import java.util.Collection; +import java.util.Iterator; +import java.util.Properties; +import java.util.UUID; + +import org.junit.Test; + +import org.springframework.context.support.ClassPathXmlApplicationContext; +import org.springframework.data.mongodb.core.SimpleMongoDbFactory; +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; +import org.springframework.integration.store.MessageGroup; +import org.springframework.integration.store.MessageGroupStore; +import org.springframework.integration.store.MessageStore; +import org.springframework.integration.store.SimpleMessageGroup; +import org.springframework.integration.support.MessageBuilder; + +import com.mongodb.Mongo; + +/** + * @author Oleg Zhurakousky + * @author Gary Russell + * @author Amol Nayak + * + */ +public abstract class AbstractMongoDbMessageGroupStoreTests extends MongoDbAvailableTests { + + @Test + @MongoDbAvailable + public void testNonExistingEmptyMessageGroup() throws Exception{ + this.cleanupCollections(new SimpleMongoDbFactory(new Mongo(), "test")); + MessageGroupStore store = getMessageGroupStore(); + MessageGroup messageGroup = store.getMessageGroup(1); + assertNotNull(messageGroup); + assertTrue(messageGroup instanceof SimpleMessageGroup); + assertEquals(0, messageGroup.size()); + } + + @Test + @MongoDbAvailable + public void testMessageGroupWithAddedMessagePrimitiveGroupId() throws Exception{ + this.cleanupCollections(new SimpleMongoDbFactory(new Mongo(), "test")); + MessageGroupStore store = this.getMessageGroupStore(); + MessageStore messageStore = this.getMessageStore(); + MessageGroup messageGroup = store.getMessageGroup(1); + Message messageA = new GenericMessage("A"); + Message messageB = new GenericMessage("B"); + store.addMessageToGroup(1, messageA); + messageGroup = store.addMessageToGroup(1, messageB); + assertNotNull(messageGroup); + assertEquals(2, messageGroup.size()); + Message retrievedMessage = messageStore.getMessage(messageA.getHeaders().getId()); + assertNotNull(retrievedMessage); + assertEquals(retrievedMessage.getHeaders().getId(), messageA.getHeaders().getId()); + // ensure that 'message_group' header that is only used internally is not propagated + assertNull(retrievedMessage.getHeaders().get("message_group")); + } + + @Test + @MongoDbAvailable + public void testMessageGroupWithAddedMessageUUIDGroupIdAndUUIDHeader() throws Exception{ + this.cleanupCollections(new SimpleMongoDbFactory(new Mongo(), "test")); + MessageGroupStore store = this.getMessageGroupStore(); + MessageStore messageStore = this.getMessageStore(); + Object id = UUID.randomUUID(); + MessageGroup messageGroup = store.getMessageGroup(id); + UUID uuidA = UUID.randomUUID(); + Message messageA = MessageBuilder.withPayload("A").setHeader("foo", uuidA).build(); + UUID uuidB = UUID.randomUUID(); + Message messageB = MessageBuilder.withPayload("B").setHeader("foo", uuidB).build(); + store.addMessageToGroup(id, messageA); + messageGroup = store.addMessageToGroup(id, messageB); + assertNotNull(messageGroup); + assertEquals(2, messageGroup.size()); + Message retrievedMessage = messageStore.getMessage(messageA.getHeaders().getId()); + assertNotNull(retrievedMessage); + assertEquals(retrievedMessage.getHeaders().getId(), messageA.getHeaders().getId()); + // ensure that 'message_group' header that is only used internally is not propagated + assertNull(retrievedMessage.getHeaders().get("message_group")); + Object fooHeader = retrievedMessage.getHeaders().get("foo"); + assertTrue(fooHeader instanceof UUID); + assertEquals(uuidA, fooHeader); + } + + @Test + @MongoDbAvailable + public void testCountMessagesInGroup() throws Exception{ + this.cleanupCollections(new SimpleMongoDbFactory(new Mongo(), "test")); + MessageGroupStore store = this.getMessageGroupStore(); + Message messageA = new GenericMessage("A"); + Message messageB = new GenericMessage("B"); + store.addMessageToGroup(1, messageA); + store.addMessageToGroup(1, messageB); + assertEquals(2, store.messageGroupSize(1)); + } + + @Test + @MongoDbAvailable + public void testPollMessages() throws Exception{ + this.cleanupCollections(new SimpleMongoDbFactory(new Mongo(), "test")); + MessageGroupStore store = this.getMessageGroupStore(); + Message messageA = new GenericMessage("A"); + Message messageB = new GenericMessage("B"); + store.addMessageToGroup(1, messageA); + store.addMessageToGroup(1, messageB); + assertEquals(2, store.messageGroupSize(1)); + Message out = store.pollMessageFromGroup(1); + assertNotNull(out); + assertEquals("A", out.getPayload()); + assertEquals(1, store.messageGroupSize(1)); + out = store.pollMessageFromGroup(1); + assertEquals("B", out.getPayload()); + assertEquals(0, store.messageGroupSize(1)); + } + + @Test + @MongoDbAvailable + public void testSameMessageMultipleGroupsPoll() throws Exception{ + this.cleanupCollections(new SimpleMongoDbFactory(new Mongo(), "test")); + MessageGroupStore store = this.getMessageGroupStore(); + Message messageA = new GenericMessage("A"); + store.addMessageToGroup(1, messageA); + store.addMessageToGroup(2, messageA); + store.addMessageToGroup(3, messageA); + store.addMessageToGroup(4, messageA); + assertEquals(1, store.messageGroupSize(1)); + assertEquals(1, store.messageGroupSize(2)); + assertEquals(1, store.messageGroupSize(3)); + assertEquals(1, store.messageGroupSize(4)); + store.pollMessageFromGroup(3); + assertEquals(1, store.messageGroupSize(1)); + assertEquals(1, store.messageGroupSize(2)); + assertEquals(0, store.messageGroupSize(3)); + assertEquals(1, store.messageGroupSize(4)); + store.pollMessageFromGroup(4); + assertEquals(1, store.messageGroupSize(1)); + assertEquals(1, store.messageGroupSize(2)); + assertEquals(0, store.messageGroupSize(3)); + assertEquals(0, store.messageGroupSize(4)); + store.pollMessageFromGroup(2); + assertEquals(1, store.messageGroupSize(1)); + assertEquals(0, store.messageGroupSize(2)); + assertEquals(0, store.messageGroupSize(3)); + assertEquals(0, store.messageGroupSize(4)); + store.pollMessageFromGroup(1); + assertEquals(0, store.messageGroupSize(1)); + assertEquals(0, store.messageGroupSize(2)); + assertEquals(0, store.messageGroupSize(3)); + assertEquals(0, store.messageGroupSize(4)); + } + + @Test + @MongoDbAvailable + public void testSameMessageMultipleGroupsRemove() throws Exception{ + this.cleanupCollections(new SimpleMongoDbFactory(new Mongo(), "test")); + MessageGroupStore store = this.getMessageGroupStore(); + + Message messageA = new GenericMessage("A"); + store.addMessageToGroup(1, messageA); + store.addMessageToGroup(2, messageA); + store.addMessageToGroup(3, messageA); + store.addMessageToGroup(4, messageA); + assertEquals(1, store.messageGroupSize(1)); + assertEquals(1, store.messageGroupSize(2)); + assertEquals(1, store.messageGroupSize(3)); + assertEquals(1, store.messageGroupSize(4)); + store.removeMessageFromGroup(3, messageA); + assertEquals(1, store.messageGroupSize(1)); + assertEquals(1, store.messageGroupSize(2)); + assertEquals(0, store.messageGroupSize(3)); + assertEquals(1, store.messageGroupSize(4)); + store.removeMessageFromGroup(4, messageA); + assertEquals(1, store.messageGroupSize(1)); + assertEquals(1, store.messageGroupSize(2)); + assertEquals(0, store.messageGroupSize(3)); + assertEquals(0, store.messageGroupSize(4)); + store.removeMessageFromGroup(2, messageA); + assertEquals(1, store.messageGroupSize(1)); + assertEquals(0, store.messageGroupSize(2)); + assertEquals(0, store.messageGroupSize(3)); + assertEquals(0, store.messageGroupSize(4)); + store.removeMessageFromGroup(1, messageA); + assertEquals(0, store.messageGroupSize(1)); + assertEquals(0, store.messageGroupSize(2)); + assertEquals(0, store.messageGroupSize(3)); + assertEquals(0, store.messageGroupSize(4)); + } + + @Test + @MongoDbAvailable + public void testMessageGroupUpdatedDateChangesWithEachAddedMessage() throws Exception{ + this.cleanupCollections(new SimpleMongoDbFactory(new Mongo(), "test")); + MessageGroupStore store = this.getMessageGroupStore(); + + MessageGroup messageGroup = store.getMessageGroup(1); + Message message = new GenericMessage("Hello"); + messageGroup = store.addMessageToGroup(1, message); + assertNotNull(messageGroup); + assertEquals(1, messageGroup.size()); + long createdTimestamp = messageGroup.getTimestamp(); + long updatedTimestamp = messageGroup.getLastModified(); + assertEquals(createdTimestamp, updatedTimestamp); + Thread.sleep(1000); + message = new GenericMessage("Hello again"); + messageGroup = store.addMessageToGroup(1, message); + createdTimestamp = messageGroup.getTimestamp(); + updatedTimestamp = messageGroup.getLastModified(); + assertTrue(updatedTimestamp > createdTimestamp); + assertEquals(2, messageGroup.size()); + + // make sure the store is properly rebuild from MongoDB + store = this.getMessageGroupStore(); + + messageGroup = store.getMessageGroup(1); + assertEquals(2, messageGroup.size()); + } + + @Test + @MongoDbAvailable + public void testMessageGroupMarkingMessage() throws Exception{ + this.cleanupCollections(new SimpleMongoDbFactory(new Mongo(), "test")); + MessageGroupStore store = this.getMessageGroupStore(); + + MessageGroup messageGroup = store.getMessageGroup(1); + Message messageA = new GenericMessage("A"); + Message messageB = new GenericMessage("B"); + store.addMessageToGroup(1, messageA); + messageGroup = store.addMessageToGroup(1, messageB); + assertNotNull(messageGroup); + assertEquals(2, messageGroup.size()); + + messageGroup = store.removeMessageFromGroup(1, messageA); + assertEquals(1, messageGroup.size()); + + // validate that the updates were propagated to Mongo as well + store = this.getMessageGroupStore(); + + messageGroup = store.getMessageGroup(1); + assertEquals(1, messageGroup.size()); + } + + @Test + @MongoDbAvailable + public void testRemoveMessageGroup() throws Exception{ + this.cleanupCollections(new SimpleMongoDbFactory(new Mongo(), "test")); + MessageGroupStore store = this.getMessageGroupStore(); + MessageStore messageStore = this.getMessageStore(); + + MessageGroup messageGroup = store.getMessageGroup(1); + Message message = new GenericMessage("Hello"); + UUID id = message.getHeaders().getId(); + messageGroup = store.addMessageToGroup(1, message); + assertNotNull(messageGroup); + assertEquals(1, messageGroup.size()); + message = messageStore.getMessage(id); + assertNotNull(message); + + store.removeMessageGroup(1); + MessageGroup messageGroupA = store.getMessageGroup(1); + assertEquals(0, messageGroupA.size()); + assertFalse(messageGroupA.equals(messageGroup)); + + } + + @Test + @MongoDbAvailable + public void testCompleteMessageGroup() throws Exception{ + this.cleanupCollections(new SimpleMongoDbFactory(new Mongo(), "test")); + MessageGroupStore store = this.getMessageGroupStore(); + + MessageGroup messageGroup = store.getMessageGroup(1); + assertNotNull(messageGroup); + Message message = new GenericMessage("Hello"); + store.addMessageToGroup(messageGroup.getGroupId(), message); + store.completeGroup(messageGroup.getGroupId()); + messageGroup = store.getMessageGroup(1); + assertTrue(messageGroup.isComplete()); + } + + @Test + @MongoDbAvailable + public void testLastReleasedSequenceNumber() throws Exception{ + this.cleanupCollections(new SimpleMongoDbFactory(new Mongo(), "test")); + MessageGroupStore store = this.getMessageGroupStore(); + + MessageGroup messageGroup = store.getMessageGroup(1); + assertNotNull(messageGroup); + Message message = new GenericMessage("Hello"); + store.addMessageToGroup(messageGroup.getGroupId(), message); + store.setLastReleasedSequenceNumberForGroup(messageGroup.getGroupId(), 5); + messageGroup = store.getMessageGroup(1); + assertEquals(5, messageGroup.getLastReleasedMessageSequenceNumber()); + } + + @Test + @MongoDbAvailable + public void testRemoveMessageFromTheGroup() throws Exception{ + this.cleanupCollections(new SimpleMongoDbFactory(new Mongo(), "test")); + MessageGroupStore store = this.getMessageGroupStore(); + + MessageGroup messageGroup = store.getMessageGroup(1); + Message message = new GenericMessage("2"); + store.addMessageToGroup(1, new GenericMessage("1")); + store.addMessageToGroup(1, message); + messageGroup = store.addMessageToGroup(1, new GenericMessage("3")); + assertNotNull(messageGroup); + assertEquals(3, messageGroup.size()); + + messageGroup = store.removeMessageFromGroup(1, message); + assertEquals(2, messageGroup.size()); + } + + @Test + @MongoDbAvailable + public void testMultipleMessageStores() throws Exception{ + + this.cleanupCollections(new SimpleMongoDbFactory(new Mongo(), "test")); + MessageGroupStore store1 = this.getMessageGroupStore(); + MessageGroupStore store2 = this.getMessageGroupStore(); + + Message message = new GenericMessage("1"); + store1.addMessageToGroup(1, message); + store2.addMessageToGroup(1, new GenericMessage("2")); + store1.addMessageToGroup(1, new GenericMessage("3")); + + MessageGroupStore store3 = this.getMessageGroupStore(); + + MessageGroup messageGroup = store3.getMessageGroup(1); + + assertNotNull(messageGroup); + assertEquals(3, messageGroup.size()); + + store3.removeMessageFromGroup(1, message); + + messageGroup = store2.getMessageGroup(1); + assertEquals(2, messageGroup.size()); + } + + @Test + @MongoDbAvailable + public void testMessageGroupIterator() throws Exception{ + this.cleanupCollections(new SimpleMongoDbFactory(new Mongo(), "test")); + MessageGroupStore store1 = this.getMessageGroupStore(); + MessageGroupStore store2 = this.getMessageGroupStore(); + + Message message = new GenericMessage("1"); + store2.addMessageToGroup(1, message); + store1.addMessageToGroup(2, new GenericMessage("2")); + store2.addMessageToGroup(3, new GenericMessage("3")); + + MessageGroupStore store3 = this.getMessageGroupStore(); + Iterator iterator = store3.iterator(); + assertNotNull(iterator); + int counter = 0; + while (iterator.hasNext()) { + iterator.next(); + counter++; + } + assertEquals(3, counter); + + store2.removeMessageFromGroup(1, message); + + iterator = store3.iterator(); + counter = 0; + while (iterator.hasNext()) { + iterator.next(); + counter++; + } + assertEquals(2, counter); + } + +// @Test +// @MongoDbAvailable +// public void testConcurrentModifications() throws Exception{ +// MongoDbFactory mongoDbFactory = this.prepareMongoFactory(); +// final MongoDbMessageStore store1 = new MongoDbMessageStore(mongoDbFactory); +// final MongoDbMessageStore store2 = new MongoDbMessageStore(mongoDbFactory); +// +// final Message message = new GenericMessage("1"); +// +// ExecutorService executor = null; +// +// final List failures = new ArrayList(); +// +// for (int i = 0; i < 100; i++) { +// executor = Executors.newCachedThreadPool(); +// +// executor.execute(new Runnable() { +// public void run() { +// MessageGroup group = store1.addMessageToGroup(1, message); +// if (group.getUnmarked().size() != 1){ +// failures.add("ADD"); +// throw new AssertionFailedError("Failed on ADD"); +// } +// } +// }); +// executor.execute(new Runnable() { +// public void run() { +// MessageGroup group = store2.removeMessageFromGroup(1, message); +// if (group.getUnmarked().size() != 0){ +// failures.add("REMOVE"); +// throw new AssertionFailedError("Failed on Remove"); +// } +// } +// }); +// +// executor.shutdown(); +// executor.awaitTermination(10, TimeUnit.SECONDS); +// store2.removeMessageFromGroup(1, message); // ensures that if ADD thread executed after REMOVE, the store is empty for the next cycle +// } +// assertTrue(failures.size() == 0); +// } + + + protected void testWithAggregatorWithShutdown(String config) throws Exception{ + this.cleanupCollections(new SimpleMongoDbFactory(new Mongo(), "test")); + + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(config, this.getClass()); + context.refresh(); + + MessageChannel input = context.getBean("inputChannel", MessageChannel.class); + QueueChannel output = context.getBean("outputChannel", QueueChannel.class); + + Message m1 = MessageBuilder.withPayload("1").setSequenceNumber(1).setSequenceSize(3).setCorrelationId(1).build(); + Message m2 = MessageBuilder.withPayload("2").setSequenceNumber(2).setSequenceSize(3).setCorrelationId(1).build(); + input.send(m1); + assertNull(output.receive(1000)); + input.send(m2); + assertNull(output.receive(1000)); + context.close(); + + context = new ClassPathXmlApplicationContext(config, this.getClass()); + input = context.getBean("inputChannel", MessageChannel.class); + output = context.getBean("outputChannel", QueueChannel.class); + + Message m3 = MessageBuilder.withPayload("3").setSequenceNumber(3).setSequenceSize(3).setCorrelationId(1).build(); + input.send(m3); + assertNotNull(output.receive(2000)); + } + + @Test + @MongoDbAvailable + public void testWithMessageHistory() throws Exception{ + this.cleanupCollections(new SimpleMongoDbFactory(new Mongo(), "test")); + MessageGroupStore store = this.getMessageGroupStore(); + + 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); + MessageGroup group = store.getMessageGroup(1); + assertNotNull(group); + Collection> messages = group.getMessages(); + assertTrue(!messages.isEmpty()); + message = messages.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")); + } + + protected abstract MessageGroupStore getMessageGroupStore() throws Exception; + + protected abstract MessageStore getMessageStore() throws Exception; + +} diff --git a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/AbstractMongoDbMessageStoreTests.java b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/AbstractMongoDbMessageStoreTests.java new file mode 100644 index 0000000000..8f150d8a77 --- /dev/null +++ b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/AbstractMongoDbMessageStoreTests.java @@ -0,0 +1,295 @@ +/* + * Copyright 2002-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.assertNull; +import static org.junit.Assert.assertTrue; + +import java.io.Serializable; +import java.util.Properties; +import java.util.UUID; + +import org.junit.Test; + +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.store.MessageStore; +import org.springframework.integration.support.MessageBuilder; + +import com.mongodb.Mongo; + +/** + * + * @author Mark Fisher + * @author Oleg Zhurakousky + * @author Artem Bilan + * @author Amol Nayak + * + */ +public abstract class AbstractMongoDbMessageStoreTests extends MongoDbAvailableTests { + + + @Test + @MongoDbAvailable + public void testAddGetWithStringPayload() throws Exception { + cleanupCollections(new SimpleMongoDbFactory(new Mongo(), "test")); + MessageStore store = getMessageStore(); + Message messageToStore = MessageBuilder.withPayload("Hello").build(); + store.addMessage(messageToStore); + Message retrievedMessage = store.getMessage(messageToStore.getHeaders().getId()); + assertNotNull(retrievedMessage); + assertEquals(messageToStore.getPayload(), retrievedMessage.getPayload()); + assertEquals(messageToStore.getHeaders(), retrievedMessage.getHeaders()); + assertEquals(messageToStore, retrievedMessage); + } + + + @Test + @MongoDbAvailable + public void testAddThenRemoveWithStringPayload() throws Exception { + cleanupCollections(new SimpleMongoDbFactory(new Mongo(), "test")); + MessageStore store = getMessageStore(); + Message messageToStore = MessageBuilder.withPayload("Hello").build(); + store.addMessage(messageToStore); + Message retrievedMessage = store.getMessage(messageToStore.getHeaders().getId()); + assertNotNull(retrievedMessage); + store.removeMessage(retrievedMessage.getHeaders().getId()); + retrievedMessage = store.getMessage(messageToStore.getHeaders().getId()); + assertNull(retrievedMessage); + } + + + @Test + @MongoDbAvailable + public void testAddGetWithObjectDefaultConstructorPayload() throws Exception { + cleanupCollections(new SimpleMongoDbFactory(new Mongo(), "test")); + MessageStore store = getMessageStore(); + Person p = new Person(); + p.setFname("John"); + p.setLname("Doe"); + Message messageToStore = MessageBuilder.withPayload(p).build(); + store.addMessage(messageToStore); + Message retrievedMessage = store.getMessage(messageToStore.getHeaders().getId()); + assertNotNull(retrievedMessage); + assertEquals(messageToStore.getPayload(), retrievedMessage.getPayload()); + assertEquals(messageToStore.getHeaders(), retrievedMessage.getHeaders()); + assertEquals(messageToStore, retrievedMessage); + } + + + @Test + @MongoDbAvailable + public void testWithMessageHistory() throws Exception{ + cleanupCollections(new SimpleMongoDbFactory(new Mongo(), "test")); + MessageStore store = getMessageStore(); + 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()); + assertNotNull(message); + 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")); + } + + + @Test + @MongoDbAvailable + public void testInt3153SequenceDetails() throws Exception{ + cleanupCollections(new SimpleMongoDbFactory(new Mongo(), "test")); + MessageStore store = getMessageStore(); + Message messageToStore = MessageBuilder.withPayload("test") + .pushSequenceDetails(UUID.randomUUID(), 1, 1) + .pushSequenceDetails(UUID.randomUUID(), 1, 1) + .build(); + store.addMessage(messageToStore); + Message retrievedMessage = store.getMessage(messageToStore.getHeaders().getId()); + assertNotNull(retrievedMessage); + assertEquals(messageToStore.getPayload(), retrievedMessage.getPayload()); + assertEquals(messageToStore.getHeaders(), retrievedMessage.getHeaders()); + assertEquals(messageToStore, retrievedMessage); + } + + public static class Foo implements Serializable { + /** + * + */ + private static final long serialVersionUID = 1L; + private String name; + + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } + } + + public static class Bar implements Serializable{ + /** + * + */ + private static final long serialVersionUID = 1L; + private final String name; + + public Bar(String name){ + this.name = name; + } + + public String getName() { + return name; + } + } + + public static class Baz implements Serializable { + /** + * + */ + private static final long serialVersionUID = 1L; + private final String name = "baz"; + + public String getName() { + return name; + } + } + + public static class Abc implements Serializable { + /** + * + */ + private static final long serialVersionUID = 1L; + + private final String name = "abx"; + + private Abc(){} + + public String getName() { + return name; + } + } + + public static class Xyz implements Serializable { + /** + * + */ + private static final long serialVersionUID = 1L; + + @SuppressWarnings("unused") + private final String name = "xyz"; + + private Xyz(){} + } + + + public static class Person implements Serializable{ + + /** + * + */ + private static final long serialVersionUID = 1L; + + private String fname; + + private String lname; + + public String getFname() { + return fname; + } + + public void setFname(String fname) { + this.fname = fname; + } + + public String getLname() { + return lname; + } + + public void setLname(String lname) { + this.lname = lname; + } + + @Override + public int hashCode() { + final int prime = 31; + int result = 1; + result = prime * result + ((fname == null) ? 0 : fname.hashCode()); + result = prime * result + ((lname == null) ? 0 : lname.hashCode()); + return result; + } + + @Override + public boolean equals(Object obj) { + if (this == obj) { + return true; + } + if (obj == null) { + return false; + } + if (getClass() != obj.getClass()) { + return false; + } + Person other = (Person) obj; + if (fname == null) { + if (other.fname != null) { + return false; + } + } + else if (!fname.equals(other.fname)) { + return false; + } + if (lname == null) { + if (other.lname != null) { + return false; + } + } + else if (!lname.equals(other.lname)) { + return false; + } + return true; + } + } + + protected abstract MessageStore getMessageStore() throws Exception; +} diff --git a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/ConfigurableMongoDbMessageGroupStoreTests.java b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/ConfigurableMongoDbMessageGroupStoreTests.java new file mode 100644 index 0000000000..922979a090 --- /dev/null +++ b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/ConfigurableMongoDbMessageGroupStoreTests.java @@ -0,0 +1,103 @@ +/* + * Copyright 2002-2012 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 java.util.Map; + +import org.junit.Test; + +import org.springframework.context.support.ClassPathXmlApplicationContext; +import org.springframework.context.support.GenericApplicationContext; +import org.springframework.core.convert.converter.Converter; +import org.springframework.data.mongodb.MongoDbFactory; +import org.springframework.data.mongodb.core.SimpleMongoDbFactory; +import org.springframework.integration.Message; +import org.springframework.integration.mongodb.rules.MongoDbAvailable; +import org.springframework.integration.store.MessageStore; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.integration.test.util.TestUtils; + +import com.mongodb.DBObject; +import com.mongodb.Mongo; + +/** + * @author Amol Nayak + * @author Artem Bilan + * + */ +public class ConfigurableMongoDbMessageGroupStoreTests extends AbstractMongoDbMessageGroupStoreTests { + + /* (non-Javadoc) + * @see org.springframework.integration.mongodb.store.AbstractMongoDbMessageGroupStoreTests#getMessageGroupStore() + */ + @Override + protected ConfigurableMongoDbMessageStore getMessageGroupStore() throws Exception { + MongoDbFactory mongoDbFactory = new SimpleMongoDbFactory(new Mongo(), "test"); + ConfigurableMongoDbMessageStore mongoDbMessageStore = new ConfigurableMongoDbMessageStore(mongoDbFactory); + GenericApplicationContext testApplicationContext = TestUtils.createTestApplicationContext(); + testApplicationContext.refresh(); + mongoDbMessageStore.setApplicationContext(testApplicationContext); + mongoDbMessageStore.afterPropertiesSet(); + return mongoDbMessageStore; + } + + /* (non-Javadoc) + * @see org.springframework.integration.mongodb.store.AbstractMongoDbMessageGroupStoreTests#getMessageStore() + */ + @Override + protected MessageStore getMessageStore() throws Exception { + return this.getMessageGroupStore(); + } + + @Test + @MongoDbAvailable + public void testWithAggregatorWithShutdown() throws Exception { + super.testWithAggregatorWithShutdown("mongo-aggregator-confugurable-config.xml"); + } + + @Test + @MongoDbAvailable + public void testWithCustomConverter() throws Exception { + this.prepareMongoFactory("testConfigurableMongoDbMessageStore"); + ClassPathXmlApplicationContext context = + new ClassPathXmlApplicationContext("ConfigurableMongoDbMessageStore-CustomConverter.xml", this.getClass()); + context.refresh(); + + TestGateway gateway = context.getBean(TestGateway.class); + String result = gateway.service("foo"); + assertEquals("FOO", result); + + } + + public static interface TestGateway { + + String service(String payload); + + } + + public static class MessageReadConverter implements Converter> { + + @Override + @SuppressWarnings("unchecked") + public Message convert(DBObject source) { + return MessageBuilder.withPayload(source.get("payload")).copyHeaders((Map) source.get("headers")).build(); + } + + } + +} diff --git a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/ConfigurableMongoDbMessageStore-CustomConverter.xml b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/ConfigurableMongoDbMessageStore-CustomConverter.xml new file mode 100644 index 0000000000..0fa44dde9a --- /dev/null +++ b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/ConfigurableMongoDbMessageStore-CustomConverter.xml @@ -0,0 +1,45 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/ConfigurableMongoDbMessageStoreTests.java b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/ConfigurableMongoDbMessageStoreTests.java new file mode 100644 index 0000000000..2a26e24e71 --- /dev/null +++ b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/ConfigurableMongoDbMessageStoreTests.java @@ -0,0 +1,123 @@ +/* + * Copyright 2002-2012 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.assertThat; +import static org.junit.Assert.assertTrue; + +import org.hamcrest.Matchers; +import org.junit.Test; + +import org.springframework.context.support.GenericApplicationContext; +import org.springframework.data.mongodb.MongoDbFactory; +import org.springframework.data.mongodb.core.SimpleMongoDbFactory; +import org.springframework.integration.Message; +import org.springframework.integration.MessagingException; +import org.springframework.integration.message.AdviceMessage; +import org.springframework.integration.message.ErrorMessage; +import org.springframework.integration.message.GenericMessage; +import org.springframework.integration.mongodb.rules.MongoDbAvailable; +import org.springframework.integration.store.MessageStore; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.integration.test.util.TestUtils; + +import com.mongodb.Mongo; + +/** + * @author Amol Nayak + * @author Artem Bilan + */ +public class ConfigurableMongoDbMessageStoreTests extends AbstractMongoDbMessageStoreTests { + + @Override + protected MessageStore getMessageStore() throws Exception { + MongoDbFactory mongoDbFactory = new SimpleMongoDbFactory(new Mongo(), "test"); + ConfigurableMongoDbMessageStore mongoDbMessageStore = new ConfigurableMongoDbMessageStore(mongoDbFactory); + GenericApplicationContext testApplicationContext = TestUtils.createTestApplicationContext(); + testApplicationContext.refresh(); + mongoDbMessageStore.setApplicationContext(testApplicationContext); + mongoDbMessageStore.afterPropertiesSet(); + return mongoDbMessageStore; + } + + + @Test + @MongoDbAvailable + public void testInt3076MessageAsPayload() throws Exception{ + MessageStore store = this.getMessageStore(); + Person p = new Person(); + p.setFname("John"); + p.setLname("Doe"); + Message messageToStore = new GenericMessage>(MessageBuilder.withPayload(p).build()); + store.addMessage(messageToStore); + Message retrievedMessage = store.getMessage(messageToStore.getHeaders().getId()); + assertNotNull(retrievedMessage); + assertTrue(retrievedMessage.getPayload() instanceof GenericMessage); + assertEquals(messageToStore.getPayload(), retrievedMessage.getPayload()); + assertEquals(messageToStore.getHeaders(), retrievedMessage.getHeaders()); + assertEquals(((Message) messageToStore.getPayload()).getPayload(), p); + assertEquals(messageToStore, retrievedMessage); + } + + @Test + @MongoDbAvailable + public void testInt3076AdviceMessage() throws Exception{ + MessageStore store = this.getMessageStore(); + Person p = new Person(); + p.setFname("John"); + p.setLname("Doe"); + Message inputMessage = MessageBuilder.withPayload(p).build(); + Message messageToStore = new AdviceMessage("foo", inputMessage); + store.addMessage(messageToStore); + Message retrievedMessage = store.getMessage(messageToStore.getHeaders().getId()); + assertNotNull(retrievedMessage); + assertTrue(retrievedMessage instanceof AdviceMessage); + assertEquals(messageToStore.getPayload(), retrievedMessage.getPayload()); + assertEquals(messageToStore.getHeaders(), retrievedMessage.getHeaders()); + assertEquals(inputMessage, ((AdviceMessage) retrievedMessage).getInputMessage()); + assertEquals(messageToStore, retrievedMessage); + } + + @Test + @MongoDbAvailable + public void testInt3076ErrorMessage() throws Exception{ + MessageStore store = this.getMessageStore(); + Person p = new Person(); + p.setFname("John"); + p.setLname("Doe"); + Message failedMessage = MessageBuilder.withPayload(p).build(); + MessagingException messagingException; + try { + throw new RuntimeException("intentional"); + } + catch (Exception e) { + messagingException = new MessagingException(failedMessage, "intentional MessagingException", e); + } + Message messageToStore = new ErrorMessage(messagingException); + store.addMessage(messageToStore); + Message retrievedMessage = store.getMessage(messageToStore.getHeaders().getId()); + assertNotNull(retrievedMessage); + assertTrue(retrievedMessage instanceof ErrorMessage); + assertThat(retrievedMessage.getPayload(), Matchers.instanceOf(MessagingException.class)); + assertEquals("intentional MessagingException", ((MessagingException) retrievedMessage.getPayload()).getMessage()); + assertEquals(failedMessage, ((MessagingException) retrievedMessage.getPayload()).getFailedMessage()); + assertEquals(messageToStore.getHeaders(), retrievedMessage.getHeaders()); + } + +} diff --git a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/DelayerHandlerRescheduleIntegrationConfigurableTests-context.xml b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/DelayerHandlerRescheduleIntegrationConfigurableTests-context.xml new file mode 100644 index 0000000000..3713864e60 --- /dev/null +++ b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/DelayerHandlerRescheduleIntegrationConfigurableTests-context.xml @@ -0,0 +1,29 @@ + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/DelayerHandlerRescheduleIntegrationTests.java b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/DelayerHandlerRescheduleIntegrationTests.java index 87a7637d26..1c95f80aca 100644 --- a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/DelayerHandlerRescheduleIntegrationTests.java +++ b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/DelayerHandlerRescheduleIntegrationTests.java @@ -23,11 +23,10 @@ import static org.junit.Assert.fail; import java.util.Iterator; import java.util.concurrent.TimeUnit; +import org.hamcrest.Matchers; import org.junit.Rule; import org.junit.Test; -import org.hamcrest.Matchers; - import org.springframework.context.support.AbstractApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.Message; @@ -56,10 +55,19 @@ public class DelayerHandlerRescheduleIntegrationTests extends MongoDbAvailableTe @Test @MongoDbAvailable + public void testWithMongoDbMessageStore() throws Exception { + this.testDelayerHandlerRescheduleWithMongoDbMessageStore("DelayerHandlerRescheduleIntegrationTests-context.xml"); + } + + @Test + @MongoDbAvailable + public void testWithConfigurableMongoDbMessageStore() throws Exception { + this.testDelayerHandlerRescheduleWithMongoDbMessageStore("DelayerHandlerRescheduleIntegrationConfigurableTests-context.xml"); + } + @SuppressWarnings("unchecked") - public void testDelayerHandlerRescheduleWithMongoDbMessageStore() throws Exception { - AbstractApplicationContext context = new ClassPathXmlApplicationContext( - "DelayerHandlerRescheduleIntegrationTests-context.xml", this.getClass()); + private void testDelayerHandlerRescheduleWithMongoDbMessageStore(String config) throws Exception { + AbstractApplicationContext context = new ClassPathXmlApplicationContext(config, this.getClass()); MessageChannel input = context.getBean("input", MessageChannel.class); MessageGroupStore messageStore = context.getBean("messageStore", MessageGroupStore.class); 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 2cf1e06c37..110ae46663 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 @@ -15,470 +15,35 @@ */ package org.springframework.integration.mongodb.store; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertFalse; -import static org.junit.Assert.assertNotNull; -import static org.junit.Assert.assertNull; -import static org.junit.Assert.assertTrue; - -import java.util.Iterator; -import java.util.Properties; -import java.util.UUID; - import org.junit.Test; -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.data.mongodb.core.SimpleMongoDbFactory; 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.SimpleMessageGroup; -import org.springframework.integration.support.MessageBuilder; +import org.springframework.integration.store.MessageStore; + +import com.mongodb.Mongo; /** * @author Oleg Zhurakousky * @author Gary Russell + * @author Artem Bilan * */ -public class MongoDbMessageGroupStoreTests extends MongoDbAvailableTests { +public class MongoDbMessageGroupStoreTests extends AbstractMongoDbMessageGroupStoreTests { - @Test - @MongoDbAvailable - public void testNonExistingEmptyMessageGroup() throws Exception{ - MongoDbFactory mongoDbFactory = this.prepareMongoFactory(); - MongoDbMessageStore store = new MongoDbMessageStore(mongoDbFactory); + @Override + protected MongoDbMessageStore getMessageGroupStore() throws Exception { + return new MongoDbMessageStore( new SimpleMongoDbFactory(new Mongo(), "test")); + } - MessageGroup messageGroup = store.getMessageGroup(1); - assertNotNull(messageGroup); - assertTrue(messageGroup instanceof SimpleMessageGroup); - assertEquals(0, messageGroup.size()); + @Override + protected MessageStore getMessageStore() throws Exception { + return this.getMessageGroupStore(); } @Test @MongoDbAvailable - public void testMessageGroupWithAddedMessagePrimitiveGroupId() throws Exception{ - MongoDbFactory mongoDbFactory = this.prepareMongoFactory(); - MongoDbMessageStore store = new MongoDbMessageStore(mongoDbFactory); - - MessageGroup messageGroup = store.getMessageGroup(1); - Message messageA = new GenericMessage("A"); - Message messageB = new GenericMessage("B"); - store.addMessageToGroup(1, messageA); - messageGroup = store.addMessageToGroup(1, messageB); - assertEquals(2, messageGroup.size()); - Message retrievedMessage = store.getMessage(messageA.getHeaders().getId()); - assertNotNull(retrievedMessage); - assertEquals(retrievedMessage.getHeaders().getId(), messageA.getHeaders().getId()); - // ensure that 'message_group' header that is only used internally is not propagated - assertNull(retrievedMessage.getHeaders().get("message_group")); + public void testWithAggregatorWithShutdown() throws Exception { + super.testWithAggregatorWithShutdown("mongo-aggregator-config.xml"); } - - @Test - @MongoDbAvailable - public void testMessageGroupWithAddedMessageUUIDGroupIdAndUUIDHeader() throws Exception{ - MongoDbFactory mongoDbFactory = this.prepareMongoFactory(); - MongoDbMessageStore store = new MongoDbMessageStore(mongoDbFactory); - Object id = UUID.randomUUID(); - MessageGroup messageGroup = store.getMessageGroup(id); - UUID uuidA = UUID.randomUUID(); - Message messageA = MessageBuilder.withPayload("A").setHeader("foo", uuidA).build(); - UUID uuidB = UUID.randomUUID(); - Message messageB = MessageBuilder.withPayload("B").setHeader("foo", uuidB).build(); - store.addMessageToGroup(id, messageA); - messageGroup = store.addMessageToGroup(id, messageB); - assertEquals(2, messageGroup.size()); - Message retrievedMessage = store.getMessage(messageA.getHeaders().getId()); - assertNotNull(retrievedMessage); - assertEquals(retrievedMessage.getHeaders().getId(), messageA.getHeaders().getId()); - // ensure that 'message_group' header that is only used internally is not propagated - assertNull(retrievedMessage.getHeaders().get("message_group")); - Object fooHeader = retrievedMessage.getHeaders().get("foo"); - assertTrue(fooHeader instanceof UUID); - assertEquals(uuidA, fooHeader); - } - - @Test - @MongoDbAvailable - public void testCountMessagesInGroup() throws Exception{ - MongoDbFactory mongoDbFactory = this.prepareMongoFactory(); - MongoDbMessageStore store = new MongoDbMessageStore(mongoDbFactory); - - Message messageA = new GenericMessage("A"); - Message messageB = new GenericMessage("B"); - store.addMessageToGroup(1, messageA); - store.addMessageToGroup(1, messageB); - assertEquals(2, store.messageGroupSize(1)); - } - - @Test - @MongoDbAvailable - public void testPollMessages() throws Exception{ - MongoDbFactory mongoDbFactory = this.prepareMongoFactory(); - MongoDbMessageStore store = new MongoDbMessageStore(mongoDbFactory); - - Message messageA = new GenericMessage("A"); - Message messageB = new GenericMessage("B"); - store.addMessageToGroup(1, messageA); - store.addMessageToGroup(1, messageB); - assertEquals(2, store.messageGroupSize(1)); - Message out = store.pollMessageFromGroup(1); - assertEquals("A", out.getPayload()); - assertEquals(1, store.messageGroupSize(1)); - out = store.pollMessageFromGroup(1); - assertEquals("B", out.getPayload()); - assertEquals(0, store.messageGroupSize(1)); - } - - @Test - @MongoDbAvailable - public void testSameMessageMultipleGroupsPoll() throws Exception{ - MongoDbFactory mongoDbFactory = this.prepareMongoFactory(); - MongoDbMessageStore store = new MongoDbMessageStore(mongoDbFactory); - - Message messageA = new GenericMessage("A"); - store.addMessageToGroup(1, messageA); - store.addMessageToGroup(2, messageA); - store.addMessageToGroup(3, messageA); - store.addMessageToGroup(4, messageA); - assertEquals(1, store.messageGroupSize(1)); - assertEquals(1, store.messageGroupSize(2)); - assertEquals(1, store.messageGroupSize(3)); - assertEquals(1, store.messageGroupSize(4)); - store.pollMessageFromGroup(3); - assertEquals(1, store.messageGroupSize(1)); - assertEquals(1, store.messageGroupSize(2)); - assertEquals(0, store.messageGroupSize(3)); - assertEquals(1, store.messageGroupSize(4)); - store.pollMessageFromGroup(4); - assertEquals(1, store.messageGroupSize(1)); - assertEquals(1, store.messageGroupSize(2)); - assertEquals(0, store.messageGroupSize(3)); - assertEquals(0, store.messageGroupSize(4)); - store.pollMessageFromGroup(2); - assertEquals(1, store.messageGroupSize(1)); - assertEquals(0, store.messageGroupSize(2)); - assertEquals(0, store.messageGroupSize(3)); - assertEquals(0, store.messageGroupSize(4)); - store.pollMessageFromGroup(1); - assertEquals(0, store.messageGroupSize(1)); - assertEquals(0, store.messageGroupSize(2)); - assertEquals(0, store.messageGroupSize(3)); - assertEquals(0, store.messageGroupSize(4)); - } - - @Test - @MongoDbAvailable - public void testSameMessageMultipleGroupsRemove() throws Exception{ - MongoDbFactory mongoDbFactory = this.prepareMongoFactory(); - MongoDbMessageStore store = new MongoDbMessageStore(mongoDbFactory); - - Message messageA = new GenericMessage("A"); - store.addMessageToGroup(1, messageA); - store.addMessageToGroup(2, messageA); - store.addMessageToGroup(3, messageA); - store.addMessageToGroup(4, messageA); - assertEquals(1, store.messageGroupSize(1)); - assertEquals(1, store.messageGroupSize(2)); - assertEquals(1, store.messageGroupSize(3)); - assertEquals(1, store.messageGroupSize(4)); - store.removeMessageFromGroup(3, messageA); - assertEquals(1, store.messageGroupSize(1)); - assertEquals(1, store.messageGroupSize(2)); - assertEquals(0, store.messageGroupSize(3)); - assertEquals(1, store.messageGroupSize(4)); - store.removeMessageFromGroup(4, messageA); - assertEquals(1, store.messageGroupSize(1)); - assertEquals(1, store.messageGroupSize(2)); - assertEquals(0, store.messageGroupSize(3)); - assertEquals(0, store.messageGroupSize(4)); - store.removeMessageFromGroup(2, messageA); - assertEquals(1, store.messageGroupSize(1)); - assertEquals(0, store.messageGroupSize(2)); - assertEquals(0, store.messageGroupSize(3)); - assertEquals(0, store.messageGroupSize(4)); - store.removeMessageFromGroup(1, messageA); - assertEquals(0, store.messageGroupSize(1)); - assertEquals(0, store.messageGroupSize(2)); - assertEquals(0, store.messageGroupSize(3)); - assertEquals(0, store.messageGroupSize(4)); - } - - @Test - @MongoDbAvailable - public void testMessageGroupUpdatedDateChangesWithEachAddedMessage() throws Exception{ - MongoDbFactory mongoDbFactory = this.prepareMongoFactory(); - MongoDbMessageStore store = new MongoDbMessageStore(mongoDbFactory); - - MessageGroup messageGroup = store.getMessageGroup(1); - Message message = new GenericMessage("Hello"); - messageGroup = store.addMessageToGroup(1, message); - assertEquals(1, messageGroup.size()); - long createdTimestamp = messageGroup.getTimestamp(); - long updatedTimestamp = messageGroup.getLastModified(); - assertEquals(createdTimestamp, updatedTimestamp); - Thread.sleep(1000); - message = new GenericMessage("Hello again"); - messageGroup = store.addMessageToGroup(1, message); - createdTimestamp = messageGroup.getTimestamp(); - updatedTimestamp = messageGroup.getLastModified(); - assertTrue(updatedTimestamp > createdTimestamp); - assertEquals(2, messageGroup.size()); - - // make sure the store is properly rebuild from MongoDB - store = new MongoDbMessageStore(mongoDbFactory); - - messageGroup = store.getMessageGroup(1); - assertEquals(2, messageGroup.size()); - } - - @Test - @MongoDbAvailable - public void testMessageGroupMarkingMessage() throws Exception{ - MongoDbFactory mongoDbFactory = this.prepareMongoFactory(); - MongoDbMessageStore store = new MongoDbMessageStore(mongoDbFactory); - - MessageGroup messageGroup = store.getMessageGroup(1); - Message messageA = new GenericMessage("A"); - Message messageB = new GenericMessage("B"); - store.addMessageToGroup(1, messageA); - messageGroup = store.addMessageToGroup(1, messageB); - assertEquals(2, messageGroup.size()); - - messageGroup = store.removeMessageFromGroup(1, messageA); - assertEquals(1, messageGroup.size()); - - // validate that the updates were propagated to Mongo as well - store = new MongoDbMessageStore(mongoDbFactory); - - messageGroup = store.getMessageGroup(1); - assertEquals(1, messageGroup.size()); - } - - @Test - @MongoDbAvailable - public void testRemoveMessageGroup() throws Exception{ - MongoDbFactory mongoDbFactory = this.prepareMongoFactory(); - MongoDbMessageStore store = new MongoDbMessageStore(mongoDbFactory); - - MessageGroup messageGroup = store.getMessageGroup(1); - Message message = new GenericMessage("Hello"); - UUID id = message.getHeaders().getId(); - messageGroup = store.addMessageToGroup(1, message); - assertEquals(1, messageGroup.size()); - message = store.getMessage(id); - assertNotNull(message); - - store.removeMessageGroup(1); - MessageGroup messageGroupA = store.getMessageGroup(1); - assertEquals(0, messageGroupA.size()); - assertFalse(messageGroupA.equals(messageGroup)); - - } - - @Test - @MongoDbAvailable - public void testCompleteMessageGroup() throws Exception{ - MongoDbFactory mongoDbFactory = this.prepareMongoFactory(); - MongoDbMessageStore store = new MongoDbMessageStore(mongoDbFactory); - - MessageGroup messageGroup = store.getMessageGroup(1); - Message message = new GenericMessage("Hello"); - store.addMessageToGroup(messageGroup.getGroupId(), message); - store.completeGroup(messageGroup.getGroupId()); - messageGroup = store.getMessageGroup(1); - assertTrue(messageGroup.isComplete()); - } - - @Test - @MongoDbAvailable - public void testLastReleasedSequenceNumber() throws Exception{ - MongoDbFactory mongoDbFactory = this.prepareMongoFactory(); - MongoDbMessageStore store = new MongoDbMessageStore(mongoDbFactory); - - MessageGroup messageGroup = store.getMessageGroup(1); - Message message = new GenericMessage("Hello"); - store.addMessageToGroup(messageGroup.getGroupId(), message); - store.setLastReleasedSequenceNumberForGroup(messageGroup.getGroupId(), 5); - messageGroup = store.getMessageGroup(1); - assertEquals(5, messageGroup.getLastReleasedMessageSequenceNumber()); - } - - @Test - @MongoDbAvailable - public void testRemoveMessageFromTheGroup() throws Exception{ - MongoDbFactory mongoDbFactory = this.prepareMongoFactory(); - MongoDbMessageStore store = new MongoDbMessageStore(mongoDbFactory); - - MessageGroup messageGroup = store.getMessageGroup(1); - Message message = new GenericMessage("2"); - store.addMessageToGroup(1, new GenericMessage("1")); - store.addMessageToGroup(1, message); - messageGroup = store.addMessageToGroup(1, new GenericMessage("3")); - - assertEquals(3, messageGroup.size()); - - messageGroup = store.removeMessageFromGroup(1, message); - assertEquals(2, messageGroup.size()); - } - - @Test - @MongoDbAvailable - public void testMultipleMessageStores() throws Exception{ - MongoDbFactory mongoDbFactory = this.prepareMongoFactory(); - MongoDbMessageStore store1 = new MongoDbMessageStore(mongoDbFactory); - MongoDbMessageStore store2 = new MongoDbMessageStore(mongoDbFactory); - - Message message = new GenericMessage("1"); - store1.addMessageToGroup(1, message); - store2.addMessageToGroup(1, new GenericMessage("2")); - store1.addMessageToGroup(1, new GenericMessage("3")); - - MongoDbMessageStore store3 = new MongoDbMessageStore(mongoDbFactory); - - MessageGroup messageGroup = store3.getMessageGroup(1); - - assertEquals(3, messageGroup.size()); - - store3.removeMessageFromGroup(1, message); - - messageGroup = store2.getMessageGroup(1); - assertEquals(2, messageGroup.size()); - } - - @Test - @MongoDbAvailable - public void testMessageGroupIterator() throws Exception{ - MongoDbFactory mongoDbFactory = this.prepareMongoFactory(); - MongoDbMessageStore store1 = new MongoDbMessageStore(mongoDbFactory); - MongoDbMessageStore store2 = new MongoDbMessageStore(mongoDbFactory); - - Message message = new GenericMessage("1"); - store2.addMessageToGroup(1, message); - store1.addMessageToGroup(2, new GenericMessage("2")); - store2.addMessageToGroup(3, new GenericMessage("3")); - - MongoDbMessageStore store3 = new MongoDbMessageStore(mongoDbFactory); - - Iterator iterator = store3.iterator(); - int counter = 0; - while (iterator.hasNext()) { - iterator.next(); - counter++; - } - assertEquals(3, counter); - - store2.removeMessageFromGroup(1, message); - - iterator = store3.iterator(); - counter = 0; - while (iterator.hasNext()) { - iterator.next(); - counter++; - } - assertEquals(2, counter); - } - -// @Test -// @MongoDbAvailable -// public void testConcurrentModifications() throws Exception{ -// MongoDbFactory mongoDbFactory = this.prepareMongoFactory(); -// final MongoDbMessageStore store1 = new MongoDbMessageStore(mongoDbFactory); -// final MongoDbMessageStore store2 = new MongoDbMessageStore(mongoDbFactory); -// -// final Message message = new GenericMessage("1"); -// -// ExecutorService executor = null; -// -// final List failures = new ArrayList(); -// -// for (int i = 0; i < 100; i++) { -// executor = Executors.newCachedThreadPool(); -// -// executor.execute(new Runnable() { -// public void run() { -// MessageGroup group = store1.addMessageToGroup(1, message); -// if (group.getUnmarked().size() != 1){ -// failures.add("ADD"); -// throw new AssertionFailedError("Failed on ADD"); -// } -// } -// }); -// executor.execute(new Runnable() { -// public void run() { -// MessageGroup group = store2.removeMessageFromGroup(1, message); -// if (group.getUnmarked().size() != 0){ -// failures.add("REMOVE"); -// throw new AssertionFailedError("Failed on Remove"); -// } -// } -// }); -// -// executor.shutdown(); -// executor.awaitTermination(10, TimeUnit.SECONDS); -// store2.removeMessageFromGroup(1, message); // ensures that if ADD thread executed after REMOVE, the store is empty for the next cycle -// } -// assertTrue(failures.size() == 0); -// } - - - @Test - @MongoDbAvailable - public void testWithAggregatorWithShutdown() throws Exception{ - this.prepareMongoFactory(); // for this test it only ensures that DB was flushed before test - - ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("mongo-aggregator-config.xml", this.getClass()); - MessageChannel input = context.getBean("inputChannel", MessageChannel.class); - QueueChannel output = context.getBean("outputChannel", QueueChannel.class); - - Message m1 = MessageBuilder.withPayload("1").setSequenceNumber(1).setSequenceSize(3).setCorrelationId(1).build(); - Message m2 = MessageBuilder.withPayload("2").setSequenceNumber(2).setSequenceSize(3).setCorrelationId(1).build(); - input.send(m1); - assertNull(output.receive(1000)); - input.send(m2); - assertNull(output.receive(1000)); - context.close(); - - context = new ClassPathXmlApplicationContext("mongo-aggregator-config.xml", this.getClass()); - input = context.getBean("inputChannel", MessageChannel.class); - output = context.getBean("outputChannel", QueueChannel.class); - - Message m3 = MessageBuilder.withPayload("3").setSequenceNumber(3).setSequenceSize(3).setCorrelationId(1).build(); - input.send(m3); - 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/MongoDbMessageStoreClaimCheckIntegrationTests.java b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/MongoDbMessageStoreClaimCheckIntegrationTests.java index c67d3bb70c..de5f3c9a0e 100644 --- a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/MongoDbMessageStoreClaimCheckIntegrationTests.java +++ b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/MongoDbMessageStoreClaimCheckIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2011 the original author or authors. + * Copyright 2002-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. @@ -16,7 +16,12 @@ package org.springframework.integration.mongodb.store; +import static org.junit.Assert.assertEquals; + +import java.io.Serializable; + import org.junit.Test; + import org.springframework.data.mongodb.MongoDbFactory; import org.springframework.data.mongodb.core.SimpleMongoDbFactory; import org.springframework.integration.Message; @@ -28,18 +33,17 @@ import org.springframework.integration.transformer.ClaimCheckOutTransformer; import com.mongodb.Mongo; -import static org.junit.Assert.assertEquals; - /** * @author Mark Fisher + * @author Artem Bilan */ -public class MongoDbMessageStoreClaimCheckIntegrationTests extends MongoDbAvailableTests{ +public class MongoDbMessageStoreClaimCheckIntegrationTests extends MongoDbAvailableTests { - @Test + @Test @MongoDbAvailable public void stringPayload() throws Exception { MongoDbFactory mongoDbFactory = new SimpleMongoDbFactory(new Mongo(), "test"); - MongoDbMessageStore messageStore = new MongoDbMessageStore(mongoDbFactory); + MongoDbMessageStore messageStore = new MongoDbMessageStore(mongoDbFactory); ClaimCheckInTransformer checkin = new ClaimCheckInTransformer(messageStore); ClaimCheckOutTransformer checkout = new ClaimCheckOutTransformer(messageStore); Message originalMessage = MessageBuilder.withPayload("test1").build(); @@ -49,12 +53,9 @@ public class MongoDbMessageStoreClaimCheckIntegrationTests extends MongoDbAvaila assertEquals(claimCheckMessage.getPayload(), checkedOutMessage.getHeaders().getId()); assertEquals(originalMessage.getPayload(), checkedOutMessage.getPayload()); assertEquals(originalMessage, checkedOutMessage); - //System.out.println("original: " + originalMessage); - //System.out.println("claimcheck: " + claimCheckMessage); - //System.out.println("checkedout: " + checkedOutMessage); } - @Test + @Test @MongoDbAvailable public void objectPayload() throws Exception { MongoDbFactory mongoDbFactory = new SimpleMongoDbFactory(new Mongo(), "test"); @@ -74,14 +75,52 @@ public class MongoDbMessageStoreClaimCheckIntegrationTests extends MongoDbAvaila assertEquals(originalMessage, checkedOutMessage); } + @Test + @MongoDbAvailable + public void stringPayloadConfigurable() throws Exception { + MongoDbFactory mongoDbFactory = new SimpleMongoDbFactory(new Mongo(), "test"); + ConfigurableMongoDbMessageStore messageStore = new ConfigurableMongoDbMessageStore(mongoDbFactory); + messageStore.afterPropertiesSet(); + ClaimCheckInTransformer checkin = new ClaimCheckInTransformer(messageStore); + ClaimCheckOutTransformer checkout = new ClaimCheckOutTransformer(messageStore); + Message originalMessage = MessageBuilder.withPayload("test1").build(); + Message claimCheckMessage = checkin.transform(originalMessage); + assertEquals(originalMessage.getHeaders().getId(), claimCheckMessage.getPayload()); + Message checkedOutMessage = checkout.transform(claimCheckMessage); + assertEquals(claimCheckMessage.getPayload(), checkedOutMessage.getHeaders().getId()); + assertEquals(originalMessage.getPayload(), checkedOutMessage.getPayload()); + assertEquals(originalMessage, checkedOutMessage); + } - @SuppressWarnings("unused") - private static class Beverage { + @Test + @MongoDbAvailable + public void objectPayloadConfigurable() throws Exception { + MongoDbFactory mongoDbFactory = new SimpleMongoDbFactory(new Mongo(), "test"); + ConfigurableMongoDbMessageStore messageStore = new ConfigurableMongoDbMessageStore(mongoDbFactory); + messageStore.afterPropertiesSet(); + ClaimCheckInTransformer checkin = new ClaimCheckInTransformer(messageStore); + ClaimCheckOutTransformer checkout = new ClaimCheckOutTransformer(messageStore); + Beverage payload = new Beverage(); + payload.setName("latte"); + payload.setShots(3); + payload.setIced(false); + Message originalMessage = MessageBuilder.withPayload(payload).build(); + Message claimCheckMessage = checkin.transform(originalMessage); + assertEquals(originalMessage.getHeaders().getId(), claimCheckMessage.getPayload()); + Message checkedOutMessage = checkout.transform(claimCheckMessage); + assertEquals(originalMessage.getPayload(), checkedOutMessage.getPayload()); + assertEquals(claimCheckMessage.getPayload(), checkedOutMessage.getHeaders().getId()); + assertEquals(originalMessage, checkedOutMessage); + } + + @SuppressWarnings("serial") + private static class Beverage implements Serializable { private String name; - private int shots; + private int shots; private boolean iced; + @SuppressWarnings("unused") public String getName() { return name; } @@ -90,6 +129,7 @@ public class MongoDbMessageStoreClaimCheckIntegrationTests extends MongoDbAvaila this.name = name; } + @SuppressWarnings("unused") public int getShots() { return shots; } @@ -98,6 +138,7 @@ public class MongoDbMessageStoreClaimCheckIntegrationTests extends MongoDbAvaila this.shots = shots; } + @SuppressWarnings("unused") public boolean isIced() { return iced; } @@ -144,6 +185,7 @@ public class MongoDbMessageStoreClaimCheckIntegrationTests extends MongoDbAvaila } return true; } + } } 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 d42ddb2b8f..1264128411 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 @@ -13,246 +13,23 @@ * 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.assertNull; -import static org.junit.Assert.assertTrue; - -import java.util.Properties; -import java.util.UUID; - -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 org.springframework.integration.store.MessageStore; import com.mongodb.Mongo; - /** * @author Mark Fisher * @author Oleg Zhurakousky * @author Artem Bilan + * */ -public class MongoDbMessageStoreTests extends MongoDbAvailableTests{ +public class MongoDbMessageStoreTests extends AbstractMongoDbMessageStoreTests { - @Test - @MongoDbAvailable - public void addGetWithStringPayload() throws Exception { - MongoDbFactory mongoDbFactory = new SimpleMongoDbFactory(new Mongo(), "test"); - MongoDbMessageStore store = new MongoDbMessageStore(mongoDbFactory); - Message messageToStore = MessageBuilder.withPayload("Hello").build(); - store.addMessage(messageToStore); - Message retrievedMessage = store.getMessage(messageToStore.getHeaders().getId()); - assertNotNull(retrievedMessage); - assertEquals(messageToStore.getPayload(), retrievedMessage.getPayload()); - assertEquals(messageToStore.getHeaders(), retrievedMessage.getHeaders()); - assertEquals(messageToStore, retrievedMessage); - } - @Test - @MongoDbAvailable - public void addThenRemoveWithStringPayload() throws Exception { - MongoDbFactory mongoDbFactory = new SimpleMongoDbFactory(new Mongo(), "test"); - MongoDbMessageStore store = new MongoDbMessageStore(mongoDbFactory); - Message messageToStore = MessageBuilder.withPayload("Hello").build(); - store.addMessage(messageToStore); - Message retrievedMessage = store.getMessage(messageToStore.getHeaders().getId()); - assertNotNull(retrievedMessage); - store.removeMessage(retrievedMessage.getHeaders().getId()); - retrievedMessage = store.getMessage(messageToStore.getHeaders().getId()); - assertNull(retrievedMessage); - } - - @Test - @MongoDbAvailable - public void addGetWithObjectDefaultConstructorPayload() throws Exception { - MongoDbFactory mongoDbFactory = new SimpleMongoDbFactory(new Mongo(), "test"); - MongoDbMessageStore store = new MongoDbMessageStore(mongoDbFactory); - Person p = new Person(); - p.setFname("John"); - p.setLname("Doe"); - Message messageToStore = MessageBuilder.withPayload(p).build(); - store.addMessage(messageToStore); - Message retrievedMessage = store.getMessage(messageToStore.getHeaders().getId()); - assertNotNull(retrievedMessage); - assertEquals(messageToStore.getPayload(), retrievedMessage.getPayload()); - 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")); - } - - @Test - @MongoDbAvailable - public void testInt3153SequenceDetails() throws Exception{ - MongoDbFactory mongoDbFactory = this.prepareMongoFactory(); - MongoDbMessageStore store = new MongoDbMessageStore(mongoDbFactory); - Message messageToStore = MessageBuilder.withPayload("test") - .pushSequenceDetails(UUID.randomUUID(), 1, 1) - .pushSequenceDetails(UUID.randomUUID(), 1, 1) - .build(); - store.addMessage(messageToStore); - Message retrievedMessage = store.getMessage(messageToStore.getHeaders().getId()); - assertNotNull(retrievedMessage); - assertEquals(messageToStore.getPayload(), retrievedMessage.getPayload()); - assertEquals(messageToStore.getHeaders(), retrievedMessage.getHeaders()); - assertEquals(messageToStore, retrievedMessage); - } - - 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 { - - private String fname; - - private String lname; - - public String getFname() { - return fname; - } - - public void setFname(String fname) { - this.fname = fname; - } - - public String getLname() { - return lname; - } - - public void setLname(String lname) { - this.lname = lname; - } - - @Override - public int hashCode() { - final int prime = 31; - int result = 1; - result = prime * result + ((fname == null) ? 0 : fname.hashCode()); - result = prime * result + ((lname == null) ? 0 : lname.hashCode()); - return result; - } - - @Override - public boolean equals(Object obj) { - if (this == obj) { - return true; - } - if (obj == null) { - return false; - } - if (getClass() != obj.getClass()) { - return false; - } - Person other = (Person) obj; - if (fname == null) { - if (other.fname != null) { - return false; - } - } - else if (!fname.equals(other.fname)) { - return false; - } - if (lname == null) { - if (other.lname != null) { - return false; - } - } - else if (!lname.equals(other.lname)) { - return false; - } - return true; - } + @Override + protected MessageStore getMessageStore() throws Exception { + return new MongoDbMessageStore(new SimpleMongoDbFactory(new Mongo(), "test")); } } diff --git a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/mongo-aggregator-confugurable-config.xml b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/mongo-aggregator-confugurable-config.xml new file mode 100644 index 0000000000..c127dc0c36 --- /dev/null +++ b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/mongo-aggregator-confugurable-config.xml @@ -0,0 +1,25 @@ + + + + + + + + + + + + + + + + + + + + + diff --git a/src/reference/docbook/mongodb.xml b/src/reference/docbook/mongodb.xml index 4ebd182c5d..fa2da4d13b 100644 --- a/src/reference/docbook/mongodb.xml +++ b/src/reference/docbook/mongodb.xml @@ -111,6 +111,33 @@ and an Aggregator. As you can see it is a simple bean configuration, and it expects a MongoDbFactory as a constructor argument. + + + MongoDbMessageStore uses custom MappingMongoConverter implementation + on background to store Messages to MongoDB documents and there is some limitations + for properties (payload and headers values) of provided Message. + For example ErrorMessage can't be converted to the MongoDB document, because it has an + Exception property, which is recursive by nature. And there is no ability to configure + some custom converters for complex domain payloads or headers values. + To achieve these capabilities, the separate MongoDB MessageStore implementation has been + introduced; see next paragraph. + + + + Spring Integration 3.0 introduced ConfigurableMongoDbMessageStore - + MessageStore and MessageGroupStore implementation. + This class can apply as one of constructor argument MongoTemplate, with which you can provide + some custom WriteConcern, for example. Another constructor requires + MappingMongoConverter, alongside with MongoDbFactory, + which allows to provide some custom conversions for Messages and their properties. + Note, by default ConfigurableMongoDbMessageStore uses standard Java serialization + to write/read Messages to/from MongoDB and relies on default values of other + properties from MongoTemplate, which is built from provided + MongoDbFactory and MappingMongoConverter. + The default name for collection of ConfigurableMongoDbMessageStore is + configurableStoreMessages. It is recommended to use this implementation for robust and flexible solutions. + The MongoDbMessageStore remains for backward compatibility and may be removed in future releases. +
@@ -275,4 +302,4 @@
- \ No newline at end of file + diff --git a/src/reference/docbook/whats-new.xml b/src/reference/docbook/whats-new.xml index bafbe44a76..b7d265dafd 100644 --- a/src/reference/docbook/whats-new.xml +++ b/src/reference/docbook/whats-new.xml @@ -177,6 +177,15 @@ a message. See for more information. +
+ MongoDB support: New ConfigurableMongoDbMessageStore + + To provide more robust and flexible implementation of MessageStore + for MongoDB support has been introduced new ConfigurableMongoDbMessageStore + component. It doesn't have backward compatibility, but it is recommended to use it for new applications. + See for more information. + +