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
This commit is contained in:
Artem Bilan
committed by
Gary Russell
Gary Russell
parent
225c0b4c9f
commit
d2bb90f7e0
@@ -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<Object> customConverters = new ArrayList<Object>();
|
||||
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 <T> Message<T> addMessage(Message<T> 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<MessageDocument> 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<Message<?>> messages = new ArrayList<Message<?>>();
|
||||
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<MessageGroup> iterator() {
|
||||
Map<Object, MessageGroup> messageGroupMap = new HashMap<Object, MessageGroup>();
|
||||
Query query = Query.query(Criteria.where(GROUP_ID).exists(true));
|
||||
query.fields().include(GROUP_ID);
|
||||
List<Map> 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<Object, byte[]> serializingConverter = new SerializingConverter();
|
||||
|
||||
private final Converter<byte[], Object> deserializingConverter = new DeserializingConverter();
|
||||
|
||||
@Override
|
||||
public Set<ConvertiblePair> getConvertibleTypes() {
|
||||
Set<ConvertiblePair> convertiblePairs = new HashSet<ConvertiblePair>();
|
||||
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);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -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> S read(Class<S> clazz, DBObject source) {
|
||||
return super.read(clazz, source);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -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<String>("A");
|
||||
Message<?> messageB = new GenericMessage<String>("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<String>("A");
|
||||
Message<?> messageB = new GenericMessage<String>("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<String>("A");
|
||||
Message<?> messageB = new GenericMessage<String>("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<String>("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<String>("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<String>("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<String>("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<String>("A");
|
||||
Message<?> messageB = new GenericMessage<String>("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<String>("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<String>("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<String>("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<String>("2");
|
||||
store.addMessageToGroup(1, new GenericMessage<String>("1"));
|
||||
store.addMessageToGroup(1, message);
|
||||
messageGroup = store.addMessageToGroup(1, new GenericMessage<String>("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<String>("1");
|
||||
store1.addMessageToGroup(1, message);
|
||||
store2.addMessageToGroup(1, new GenericMessage<String>("2"));
|
||||
store1.addMessageToGroup(1, new GenericMessage<String>("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<String>("1");
|
||||
store2.addMessageToGroup(1, message);
|
||||
store1.addMessageToGroup(2, new GenericMessage<String>("2"));
|
||||
store2.addMessageToGroup(3, new GenericMessage<String>("3"));
|
||||
|
||||
MessageGroupStore store3 = this.getMessageGroupStore();
|
||||
Iterator<MessageGroup> 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<String>("1");
|
||||
//
|
||||
// ExecutorService executor = null;
|
||||
//
|
||||
// final List<Object> failures = new ArrayList<Object>();
|
||||
//
|
||||
// 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<String>("Hello");
|
||||
DirectChannel fooChannel = new DirectChannel();
|
||||
fooChannel.setBeanName("fooChannel");
|
||||
DirectChannel barChannel = new DirectChannel();
|
||||
barChannel.setBeanName("barChannel");
|
||||
|
||||
message = MessageHistory.write(message, fooChannel);
|
||||
message = MessageHistory.write(message, barChannel);
|
||||
store.addMessageToGroup(1, message);
|
||||
MessageGroup group = store.getMessageGroup(1);
|
||||
assertNotNull(group);
|
||||
Collection<Message<?>> 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;
|
||||
|
||||
}
|
||||
@@ -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;
|
||||
}
|
||||
@@ -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<DBObject, Message<?>> {
|
||||
|
||||
@Override
|
||||
@SuppressWarnings("unchecked")
|
||||
public Message<?> convert(DBObject source) {
|
||||
return MessageBuilder.withPayload(source.get("payload")).copyHeaders((Map<String,?>) source.get("headers")).build();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,45 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<beans:beans xmlns="http://www.springframework.org/schema/integration"
|
||||
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xmlns:beans="http://www.springframework.org/schema/beans"
|
||||
xmlns:mongo="http://www.springframework.org/schema/data/mongo"
|
||||
xsi:schemaLocation="http://www.springframework.org/schema/beans
|
||||
http://www.springframework.org/schema/beans/spring-beans.xsd
|
||||
http://www.springframework.org/schema/integration
|
||||
http://www.springframework.org/schema/integration/spring-integration.xsd
|
||||
http://www.springframework.org/schema/data/mongo
|
||||
http://www.springframework.org/schema/data/mongo/spring-mongo.xsd">
|
||||
|
||||
<mongo:db-factory dbname="test"/>
|
||||
|
||||
<beans:bean id="messageStore" class="org.springframework.integration.mongodb.store.ConfigurableMongoDbMessageStore">
|
||||
<beans:constructor-arg ref="mongoDbFactory"/>
|
||||
<beans:constructor-arg>
|
||||
<mongo:mapping-converter>
|
||||
<mongo:custom-converters>
|
||||
<mongo:converter>
|
||||
<beans:bean class="org.springframework.integration.mongodb.store.ConfigurableMongoDbMessageGroupStoreTests$MessageReadConverter"/>
|
||||
</mongo:converter>
|
||||
</mongo:custom-converters>
|
||||
</mongo:mapping-converter>
|
||||
</beans:constructor-arg>
|
||||
<beans:constructor-arg value="testConfigurableMongoDbMessageStore"/>
|
||||
</beans:bean>
|
||||
|
||||
<gateway id="gateway"
|
||||
service-interface="org.springframework.integration.mongodb.store.ConfigurableMongoDbMessageGroupStoreTests$TestGateway"
|
||||
default-request-channel="preInputChannel"/>
|
||||
|
||||
<header-enricher input-channel="preInputChannel" output-channel="inputChannel">
|
||||
<header-channels-to-string/>
|
||||
</header-enricher>
|
||||
|
||||
<channel id="inputChannel">
|
||||
<queue message-store="messageStore"/>
|
||||
</channel>
|
||||
|
||||
<transformer input-channel="inputChannel" expression="payload.toUpperCase()">
|
||||
<poller fixed-delay="1000"/>
|
||||
</transformer>
|
||||
|
||||
</beans:beans>
|
||||
@@ -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<Message<?>>(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<Person> 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<Person> 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());
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,29 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<beans:beans xmlns:beans="http://www.springframework.org/schema/beans"
|
||||
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xmlns="http://www.springframework.org/schema/integration"
|
||||
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
|
||||
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd">
|
||||
|
||||
<beans:bean id="mongoConnectionFactory" class="org.springframework.data.mongodb.core.SimpleMongoDbFactory">
|
||||
<beans:constructor-arg>
|
||||
<beans:bean class="com.mongodb.Mongo"/>
|
||||
</beans:constructor-arg>
|
||||
<beans:constructor-arg value="test"/>
|
||||
</beans:bean>
|
||||
|
||||
<beans:bean id="messageStore" class="org.springframework.integration.mongodb.store.ConfigurableMongoDbMessageStore">
|
||||
<beans:constructor-arg ref="mongoConnectionFactory"/>
|
||||
</beans:bean>
|
||||
|
||||
<channel id="output">
|
||||
<queue/>
|
||||
</channel>
|
||||
|
||||
<delayer id="#{T (org.springframework.integration.mongodb.store.DelayerHandlerRescheduleIntegrationTests).DELAYER_ID}"
|
||||
input-channel="input"
|
||||
output-channel="output"
|
||||
default-delay="10000"
|
||||
message-store="messageStore"/>
|
||||
|
||||
</beans:beans>
|
||||
@@ -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);
|
||||
|
||||
|
||||
@@ -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<String>("A");
|
||||
Message<?> messageB = new GenericMessage<String>("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<String>("A");
|
||||
Message<?> messageB = new GenericMessage<String>("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<String>("A");
|
||||
Message<?> messageB = new GenericMessage<String>("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<String>("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<String>("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<String>("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<String>("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<String>("A");
|
||||
Message<?> messageB = new GenericMessage<String>("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<String>("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<String>("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<String>("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<String>("2");
|
||||
store.addMessageToGroup(1, new GenericMessage<String>("1"));
|
||||
store.addMessageToGroup(1, message);
|
||||
messageGroup = store.addMessageToGroup(1, new GenericMessage<String>("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<String>("1");
|
||||
store1.addMessageToGroup(1, message);
|
||||
store2.addMessageToGroup(1, new GenericMessage<String>("2"));
|
||||
store1.addMessageToGroup(1, new GenericMessage<String>("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<String>("1");
|
||||
store2.addMessageToGroup(1, message);
|
||||
store1.addMessageToGroup(2, new GenericMessage<String>("2"));
|
||||
store2.addMessageToGroup(3, new GenericMessage<String>("3"));
|
||||
|
||||
MongoDbMessageStore store3 = new MongoDbMessageStore(mongoDbFactory);
|
||||
|
||||
Iterator<MessageGroup> 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<String>("1");
|
||||
//
|
||||
// ExecutorService executor = null;
|
||||
//
|
||||
// final List<Object> failures = new ArrayList<Object>();
|
||||
//
|
||||
// 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<String>("Hello");
|
||||
DirectChannel fooChannel = new DirectChannel();
|
||||
fooChannel.setBeanName("fooChannel");
|
||||
DirectChannel barChannel = new DirectChannel();
|
||||
barChannel.setBeanName("barChannel");
|
||||
|
||||
message = MessageHistory.write(message, fooChannel);
|
||||
message = MessageHistory.write(message, barChannel);
|
||||
store.addMessageToGroup(1, message);
|
||||
|
||||
message = store.getMessageGroup(1).getMessages().iterator().next();
|
||||
|
||||
MessageHistory messageHistory = MessageHistory.read(message);
|
||||
assertNotNull(messageHistory);
|
||||
assertEquals(2, messageHistory.size());
|
||||
Properties fooChannelHistory = messageHistory.get(0);
|
||||
assertEquals("fooChannel", fooChannelHistory.get("name"));
|
||||
assertEquals("channel", fooChannelHistory.get("type"));
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -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"));
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -0,0 +1,25 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<beans xmlns="http://www.springframework.org/schema/beans"
|
||||
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xmlns:int="http://www.springframework.org/schema/integration"
|
||||
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
|
||||
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd">
|
||||
|
||||
<int:aggregator input-channel="inputChannel" output-channel="outputChannel" message-store="mongoStore"/>
|
||||
|
||||
<int:channel id="outputChannel">
|
||||
<int:queue/>
|
||||
</int:channel>
|
||||
|
||||
<bean id="mongoStore" class="org.springframework.integration.mongodb.store.ConfigurableMongoDbMessageStore">
|
||||
<constructor-arg ref="mongoConnectionFactory"/>
|
||||
</bean>
|
||||
|
||||
<bean id="mongoConnectionFactory" class="org.springframework.data.mongodb.core.SimpleMongoDbFactory">
|
||||
<constructor-arg>
|
||||
<bean class="com.mongodb.Mongo"/>
|
||||
</constructor-arg>
|
||||
<constructor-arg value="test"/>
|
||||
</bean>
|
||||
|
||||
</beans>
|
||||
@@ -111,6 +111,33 @@
|
||||
and an <emphasis>Aggregator</emphasis>. As you can see it is a simple bean configuration, and it expects a
|
||||
<classname>MongoDbFactory</classname> as a constructor argument.
|
||||
</para>
|
||||
<important>
|
||||
<para>
|
||||
<classname>MongoDbMessageStore</classname> uses custom <classname>MappingMongoConverter</classname> implementation
|
||||
on background to store <interfacename>Message</interfacename>s to MongoDB documents and there is some limitations
|
||||
for properties (<code>payload</code> and <code>headers</code> values) of provided <interfacename>Message</interfacename>.
|
||||
For example <classname>ErrorMessage</classname> can't be converted to the MongoDB document, because it has an
|
||||
<classname>Exception</classname> property, which is recursive by nature. And there is no ability to configure
|
||||
some custom converters for complex domain <code>payload</code>s or <code>headers</code> values.
|
||||
To achieve these capabilities, the separate MongoDB <interfacename>MessageStore</interfacename> implementation has been
|
||||
introduced; see next paragraph.
|
||||
</para>
|
||||
</important>
|
||||
<para>
|
||||
<emphasis>Spring Integration 3.0</emphasis> introduced <classname>ConfigurableMongoDbMessageStore</classname> -
|
||||
<interfacename>MessageStore</interfacename> and <interfacename>MessageGroupStore</interfacename> implementation.
|
||||
This class can apply as one of constructor argument <classname>MongoTemplate</classname>, with which you can provide
|
||||
some custom <classname>WriteConcern</classname>, for example. Another constructor requires
|
||||
<classname>MappingMongoConverter</classname>, alongside with <interfacename>MongoDbFactory</interfacename>,
|
||||
which allows to provide some custom conversions for <interfacename>Message</interfacename>s and their properties.
|
||||
Note, by default <classname>ConfigurableMongoDbMessageStore</classname> uses standard Java serialization
|
||||
to write/read <interfacename>Message</interfacename>s to/from MongoDB and relies on default values of other
|
||||
properties from <classname>MongoTemplate</classname>, which is built from provided
|
||||
<interfacename>MongoDbFactory</interfacename> and <classname>MappingMongoConverter</classname>.
|
||||
The default name for collection of <classname>ConfigurableMongoDbMessageStore</classname> is
|
||||
<code>configurableStoreMessages</code>. It is recommended to use this implementation for robust and flexible solutions.
|
||||
The <classname>MongoDbMessageStore</classname> remains for backward compatibility and may be removed in future releases.
|
||||
</para>
|
||||
</section>
|
||||
|
||||
<section id="mongodb-inbound-channel-adapter">
|
||||
@@ -275,4 +302,4 @@
|
||||
</para>
|
||||
</section>
|
||||
|
||||
</chapter>
|
||||
</chapter>
|
||||
|
||||
@@ -177,6 +177,15 @@
|
||||
a message. See <xref linkend="header-enricher"/> for more information.
|
||||
</para>
|
||||
</section>
|
||||
<section id="3.0-configurable-mongo-MS">
|
||||
<title>MongoDB support: New ConfigurableMongoDbMessageStore</title>
|
||||
<para>
|
||||
To provide more robust and flexible implementation of <interfacename>MessageStore</interfacename>
|
||||
for MongoDB support has been introduced new <classname>ConfigurableMongoDbMessageStore</classname>
|
||||
component. It doesn't have backward compatibility, but it is recommended to use it for new applications.
|
||||
See <xref linkend="mongodb"/> for more information.
|
||||
</para>
|
||||
</section>
|
||||
</section>
|
||||
|
||||
<section id="3.0-general">
|
||||
|
||||
Reference in New Issue
Block a user