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

INT-3338: MongoMS: Add priority and sequence

Browse Source 
JIRA: https://jira.spring.io/browse/INT-3338

INT-3338: Add Docs

INT-3338: Make `priority` 'smart'

Add `Sort` for all group queries dependently of `priorityEnabled`.
Avoids the need to configure separate collections for different type of `MessageStore`

INT-3338: Add `MongoDbChannelMessageStore`

INT-3338: Polishing, Fixes, Improvements

Doc Polishing

Fix stream test.
This commit is contained in:
Artem Bilan
2014-04-03 16:34:23 +03:00
committed by Gary Russell
parent a7489909d7
commit 2ee179a891
17 changed files with 1149 additions and 436 deletions
Download Patch File Download Diff File Expand all files Collapse all files

255
spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/AbstractConfigurableMongoDbMessageStore.java Normal file
View File

@@ -0,0 +1,255 @@
/*
* Copyright 2014 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.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import com.mongodb.DB;
import com.mongodb.MongoException;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
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.dao.DataAccessException;
import org.springframework.data.mongodb.MongoDbFactory;
import org.springframework.data.mongodb.core.DbCallback;
import org.springframework.data.mongodb.core.FindAndModifyOptions;
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.context.IntegrationContextUtils;
import org.springframework.integration.store.BasicMessageGroupStore;
import org.springframework.integration.support.DefaultMessageBuilderFactory;
import org.springframework.integration.support.MessageBuilderFactory;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.util.Assert;
/**
* The abstract MongoDB {@link BasicMessageGroupStore} implementation to provide configuration for common options
* for implementations of this class.
*
* @author Artem Bilan
* @since 4.0
*/
public abstract class AbstractConfigurableMongoDbMessageStore implements BasicMessageGroupStore, InitializingBean,
ApplicationContextAware {
public final static String SEQUENCE_NAME = "messagesSequence";
/**
* 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 = "MongoDbMessageStore.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 = "MongoDbMessageStore.CREATED_DATE";
protected final Log logger = LogFactory.getLog(this.getClass());
protected final String collectionName;
protected final MongoDbFactory mongoDbFactory;
protected MongoTemplate mongoTemplate;
protected MappingMongoConverter mappingMongoConverter;
protected ApplicationContext applicationContext;
protected MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory();
public AbstractConfigurableMongoDbMessageStore(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 AbstractConfigurableMongoDbMessageStore(MongoDbFactory mongoDbFactory, String collectionName) {
this(mongoDbFactory, null, collectionName);
}
public AbstractConfigurableMongoDbMessageStore(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;
this.messageBuilderFactory = IntegrationContextUtils.getMessageBuilderFactory(this.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(MessageDocumentFields.MESSAGE_ID, Order.ASCENDING));
indexOperations.ensureIndex(new Index(MessageDocumentFields.GROUP_ID, Order.ASCENDING)
.on(MessageDocumentFields.LAST_MODIFIED_TIME, Order.DESCENDING)
.on(MessageDocumentFields.SEQUENCE, Order.DESCENDING));
}
public Message<?> getMessage(UUID id) {
Assert.notNull(id, "'id' must not be null");
Query query = Query.query(Criteria.where(MessageDocumentFields.MESSAGE_ID).is(id));
MessageDocument document = this.mongoTemplate.findOne(query, MessageDocument.class, this.collectionName);
return document != null ? document.getMessage() : null;
}
@Override
public void removeMessageGroup(Object groupId) {
this.mongoTemplate.remove(groupIdQuery(groupId), this.collectionName);
}
@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;
}
/**
* Perform MongoDB {@code INC} operation for the document, which contains the {@link MessageDocument}
* {@code sequence}, and return the new incremented value for the new {@link MessageDocument}.
* The {@link #SEQUENCE_NAME} document is created on demand.
* @return the next sequence value.
*/
protected int getNextId() {
Query query = Query.query(Criteria.where("_id").is(SEQUENCE_NAME));
query.fields().include(MessageDocumentFields.SEQUENCE);
return (Integer) this.mongoTemplate.findAndModify(query,
new Update().inc(MessageDocumentFields.SEQUENCE, 1),
FindAndModifyOptions.options().returnNew(true).upsert(true),
Map.class, this.collectionName)
.get(MessageDocumentFields.SEQUENCE);
}
protected void addMessageDocument(final MessageDocument document) {
this.mongoTemplate.executeInSession(new DbCallback<Void>() {
@Override
public Void doInDB(DB db) throws MongoException, DataAccessException {
Message<?> message = document.getMessage();
if (message.getHeaders().containsKey(SAVED_KEY)) {
Message<?> saved = getMessage(message.getHeaders().getId());
if (saved != null) {
if (saved.equals(message)) {
return null;
} // We need to save it under its own id
}
}
final long createdDate = document.getCreatedTime() == 0 ? System.currentTimeMillis() : document.getCreatedTime();
Message<?> result = messageBuilderFactory.fromMessage(message).setHeader(SAVED_KEY, Boolean.TRUE)
.setHeader(CREATED_DATE_KEY, createdDate).build();
@SuppressWarnings("unchecked")
Map<String, Object> innerMap = (Map<String, Object>) 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);
mongoTemplate.insert(document, collectionName);
return null;
}
});
}
protected static Query groupIdQuery(Object groupId) {
return Query.query(Criteria.where(MessageDocumentFields.GROUP_ID).is(groupId));
}
/**
* A {@link org.springframework.core.convert.converter.GenericConverter} implementation to convert {@link org.springframework.messaging.Message} to
* serialized {@link byte[]} to store {@link org.springframework.messaging.Message} to the MongoDB.
* And vice versa - to convert {@link byte[]} from the MongoDB to the {@link org.springframework.messaging.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);
}
}
}
}

490
spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/ConfigurableMongoDbMessageStore.java
View File

@@ -13,47 +13,34 @@
* 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.Collection;
import java.util.Iterator;
import java.util.LinkedHashSet;
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.annotation.Id;
import com.mongodb.DB;
import com.mongodb.MongoException;
import org.springframework.dao.DataAccessException;
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.DbCallback;
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.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.jmx.export.annotation.ManagedAttribute;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.util.Assert;
/**
@@ -67,41 +54,14 @@ import org.springframework.util.Assert;
* @author Artem Bilan
* @since 3.0
*/
public class ConfigurableMongoDbMessageStore extends AbstractMessageGroupStore
implements MessageStore, InitializingBean, ApplicationContextAware {
public class ConfigurableMongoDbMessageStore extends AbstractConfigurableMongoDbMessageStore
implements MessageStore, MessageGroupStore, Iterable<MessageGroup> {
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";
private final Collection<MessageGroupCallback> expiryCallbacks = new LinkedHashSet<MessageGroupCallback>();
/**
* 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;
private volatile boolean timeoutOnIdle;
public ConfigurableMongoDbMessageStore(MongoTemplate mongoTemplate) {
@@ -109,11 +69,7 @@ public class ConfigurableMongoDbMessageStore extends AbstractMessageGroupStore
}
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;
super(mongoTemplate, collectionName);
}
public ConfigurableMongoDbMessageStore(MongoDbFactory mongoDbFactory) {
@@ -129,47 +85,35 @@ public class ConfigurableMongoDbMessageStore extends AbstractMessageGroupStore
}
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;
super(mongoDbFactory, mappingMongoConverter, collectionName);
}
@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);
}
/**
* Convenient injection point for expiry callbacks in the message store. Each of the callbacks provided will simply
* be registered with the store using {@link #registerMessageGroupExpiryCallback(MessageGroupCallback)}.
*
* @param expiryCallbacks the expiry callbacks to add
*/
public void setExpiryCallbacks(Collection<MessageGroupCallback> expiryCallbacks) {
for (MessageGroupCallback callback : expiryCallbacks) {
registerMessageGroupExpiryCallback(callback);
}
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));
}
public boolean isTimeoutOnIdle() {
return timeoutOnIdle;
}
@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;
/**
* Allows you to override the rule for the timeout calculation. Typical timeout is based from the time
* the {@link MessageGroup} was created. If you want the timeout to be based on the time
* the {@link MessageGroup} was idling (e.g., inactive from the last update) invoke this method with 'true'.
* Default is 'false'.
*
* @param timeoutOnIdle The boolean.
*/
public void setTimeoutOnIdle(boolean timeoutOnIdle) {
this.timeoutOnIdle = timeoutOnIdle;
}
@Override
@@ -179,56 +123,28 @@ public class ConfigurableMongoDbMessageStore extends AbstractMessageGroupStore
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 = this.getMessageBuilderFactory().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);
Query query = Query.query(Criteria.where(MessageDocumentFields.MESSAGE_ID).is(id));
MessageDocument document = this.mongoTemplate.findAndRemove(query, MessageDocument.class, this.collectionName);
return (document != null) ? document.getMessage() : null;
}
@Override
public long getMessageCount() {
return this.mongoTemplate.getCollection(this.collectionName).getCount();
Query query = Query.query(Criteria.where(MessageDocumentFields.MESSAGE_ID).exists(true)
.and(MessageDocumentFields.GROUP_ID).exists(false));
return this.mongoTemplate.getCollection(this.collectionName).count(query.getQueryObject());
}
@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,
Assert.notNull(groupId, "'groupId' must not be null");
Query query = groupOrderQuery(groupId);
List<MessageDocument> messageDocuments = this.mongoTemplate.find(query, MessageDocument.class,
this.collectionName);
long createdTime = 0;
@@ -256,209 +172,189 @@ public class ConfigurableMongoDbMessageStore extends AbstractMessageGroupStore
}
@Override
public MessageGroup addMessageToGroup(Object groupId, Message<?> message) {
public MessageGroup addMessageToGroup(final Object groupId, final 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;
return this.mongoTemplate.executeInSession(new DbCallback<MessageGroup>() {
if (messageDocument != null) {
createdTime = messageDocument.getCreatedTime();
lastReleasedSequence = messageDocument.getLastReleasedSequence();
complete = messageDocument.isComplete();
}
@Override
public MessageGroup doInDB(DB db) throws MongoException, DataAccessException {
Query query = groupOrderQuery(groupId);
MessageDocument messageDocument = mongoTemplate.findOne(query, MessageDocument.class, collectionName);
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());
long createdTime = 0;
int lastReleasedSequence = 0;
boolean complete = false;
this.addMessageDocument(document);
if (messageDocument != null) {
createdTime = messageDocument.getCreatedTime();
lastReleasedSequence = messageDocument.getLastReleasedSequence();
complete = messageDocument.isComplete();
}
return this.getMessageGroup(groupId);
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());
document.setSequence(getNextId());
addMessageDocument(document);
return getMessageGroup(groupId);
}
});
}
@Override
public MessageGroup removeMessageFromGroup(Object groupId, Message<?> messageToRemove) {
public MessageGroup removeMessageFromGroup(final Object groupId, final 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);
}
return this.mongoTemplate.executeInSession(new DbCallback<MessageGroup>() {
@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));
@Override
public MessageGroup doInDB(DB db) throws MongoException, DataAccessException {
Query query = groupIdQuery(groupId)
.addCriteria(Criteria.where(MessageDocumentFields.MESSAGE_ID).is(messageToRemove.getHeaders().getId()));
mongoTemplate.remove(query, collectionName);
updateGroup(groupId, lastModifiedUpdate());
return getMessageGroup(groupId);
}
}
return messageGroupMap.values().iterator();
});
}
@Override
public Message<?> pollMessageFromGroup(Object groupId) {
public Message<?> pollMessageFromGroup(final 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;
return this.mongoTemplate.executeInSession(new DbCallback<Message<?>>() {
@Override
public Message<?> doInDB(DB db) throws MongoException, DataAccessException {
Sort sort = new Sort(MessageDocumentFields.LAST_MODIFIED_TIME, MessageDocumentFields.SEQUENCE);
Query query = groupIdQuery(groupId).with(sort);
MessageDocument document = mongoTemplate.findAndRemove(query, MessageDocument.class, collectionName);
Message<?> message = null;
if (document != null) {
message = document.getMessage();
updateGroup(groupId, lastModifiedUpdate());
}
return message;
}
});
}
@Override
public void setLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber) {
this.updateGroup(groupId, lastModifiedUpdate().set(LAST_RELEASED_SEQUENCE, sequenceNumber));
this.updateGroup(groupId, lastModifiedUpdate().set(MessageDocumentFields.LAST_RELEASED_SEQUENCE, sequenceNumber));
}
@Override
public void completeGroup(Object groupId) {
this.updateGroup(groupId, lastModifiedUpdate().set(COMPLETE, true));
this.updateGroup(groupId, lastModifiedUpdate().set(MessageDocumentFields.COMPLETE, true));
}
@Override
public Iterator<MessageGroup> iterator() {
return this.mongoTemplate.executeInSession(new DbCallback<Iterator<MessageGroup>>() {
@Override
public Iterator<MessageGroup> doInDB(DB db) throws MongoException, DataAccessException {
List<MessageGroup> messageGroups = new ArrayList<MessageGroup>();
Query query = Query.query(Criteria.where(MessageDocumentFields.GROUP_ID).exists(true));
@SuppressWarnings("rawtypes")
List groupIds = mongoTemplate.getCollection(collectionName)
.distinct(MessageDocumentFields.GROUP_ID, query.getQueryObject());
for (Object groupId : groupIds) {
messageGroups.add(getMessageGroup(groupId));
}
return messageGroups.iterator();
}
});
}
@Override
public void registerMessageGroupExpiryCallback(MessageGroupCallback callback) {
expiryCallbacks.add(callback);
}
@Override
public int expireMessageGroups(long timeout) {
int count = 0;
long threshold = System.currentTimeMillis() - timeout;
for (MessageGroup group : this) {
long timestamp = group.getTimestamp();
if (this.isTimeoutOnIdle() && group.getLastModified() > 0) {
timestamp = group.getLastModified();
}
if (timestamp <= threshold) {
count++;
expire(group);
}
}
return count;
}
@Override
@ManagedAttribute
public int getMessageCountForAllMessageGroups() {
Query query = Query.query(Criteria.where(MessageDocumentFields.MESSAGE_ID).exists(true)
.and(MessageDocumentFields.GROUP_ID).exists(true));
long count = this.mongoTemplate.count(query, this.collectionName);
Assert.isTrue(count <= Integer.MAX_VALUE, "Message count is out of Integer's range");
return (int) count;
}
@Override
@ManagedAttribute
public int getMessageGroupCount() {
Query query = Query.query(Criteria.where(MessageDocumentFields.GROUP_ID).exists(true));
return this.mongoTemplate.getCollection(this.collectionName)
.distinct(MessageDocumentFields.GROUP_ID, query.getQueryObject())
.size();
}
private void expire(MessageGroup group) {
RuntimeException exception = null;
for (MessageGroupCallback callback : expiryCallbacks) {
try {
callback.execute(this, group);
}
catch (RuntimeException e) {
if (exception == null) {
exception = e;
}
logger.error("Exception in expiry callback", e);
}
}
if (exception != null) {
throw exception;
}
}
private void updateGroup(Object groupId, Update update) {
this.mongoTemplate.updateFirst(groupIdQuery(groupId), update, this.collectionName);
this.mongoTemplate.updateFirst(groupOrderQuery(groupId), update, this.collectionName);
}
private static Update lastModifiedUpdate() {
return Update.update(LAST_MODIFIED_TIME, System.currentTimeMillis());
return Update.update(MessageDocumentFields.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 {
/*
* Needed as a persistence property to suppress 'Cannot determine IsNewStrategy' MappingException
* when the application context is configured with auditing. The document is not
* currently Auditable.
*/
@SuppressWarnings("unused")
@Id
private String _id;
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);
}
}
private static Query groupOrderQuery(Object groupId) {
Sort sort = new Sort(Sort.Direction.DESC, MessageDocumentFields.LAST_MODIFIED_TIME, MessageDocumentFields.SEQUENCE);
return groupIdQuery(groupId).with(sort);
}
}

120
spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MessageDocument.java Normal file
View File

@@ -0,0 +1,120 @@
/*
* Copyright 2014 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.UUID;
import org.springframework.data.annotation.Id;
import org.springframework.data.mongodb.core.mapping.Document;
import org.springframework.messaging.Message;
import org.springframework.util.Assert;
/**
* The entity class to wrap {@link org.springframework.messaging.Message} to the MongoDB document.
*
* @author Artem Bilan
* @since 4.0
*/
@Document
public class MessageDocument {
/*
* Needed as a persistence property to suppress 'Cannot determine IsNewStrategy' MappingException
* when the application context is configured with auditing. The document is not
* currently Auditable.
*/
@SuppressWarnings("unused")
@Id
private String _id;
private final Message<?> message;
@SuppressWarnings("unused")
private final UUID messageId;
@SuppressWarnings("unused")
private Integer priority;
private Long createdTime = 0L;
@SuppressWarnings("unused")
private Object groupId;
private Long lastModifiedTime = 0L;
private Boolean complete = false;
private Integer lastReleasedSequence = 0;
@SuppressWarnings("unused")
private int sequence;
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 void setPriority(Integer priority) {
this.priority = priority;
}
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;
}
public void setSequence(int sequence) {
this.sequence = sequence;
}
}

43
spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MessageDocumentFields.java Normal file
View File

@@ -0,0 +1,43 @@
/*
* Copyright 2014 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;
/**
* @author Artem Bilan
*
* @since 4.0
*/
public final class MessageDocumentFields {
public static final String MESSAGE_ID = "messageId";
public static final String PRIORITY = "priority";
public static final String GROUP_ID = "groupId";
public static final String LAST_MODIFIED_TIME = "lastModifiedTime";
public static final String SEQUENCE = "sequence";
public static final String LAST_RELEASED_SEQUENCE = "lastReleasedSequence";
public static final String COMPLETE = "complete";
private MessageDocumentFields() {
}
}

143
spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoDbChannelMessageStore.java Normal file
View File

@@ -0,0 +1,143 @@
/*
* Copyright 2014 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 org.springframework.data.domain.Sort;
import org.springframework.data.mongodb.MongoDbFactory;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.convert.MappingMongoConverter;
import org.springframework.data.mongodb.core.index.Index;
import org.springframework.data.mongodb.core.query.Order;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.integration.IntegrationMessageHeaderAccessor;
import org.springframework.integration.store.MessageGroup;
import org.springframework.integration.store.PriorityCapableChannelMessageStore;
import org.springframework.integration.store.SimpleMessageGroup;
import org.springframework.messaging.Message;
import org.springframework.util.Assert;
/**
* MongoDB {@link PriorityCapableChannelMessageStore} implementation.
* This message store shall be used for message channels only.
*
* <p>Provide the {@link #priorityEnabled} option to allow to poll messages via {@code priority} manner.
*
* <p>As a priority document field the {@link org.springframework.integration.IntegrationMessageHeaderAccessor#PRIORITY}
* message header is used.
*
* <p>The same collection can be used for {@link org.springframework.integration.channel.QueueChannel}s and
* {@link org.springframework.integration.channel.PriorityChannel}s, but the different instances of
* {@link MongoDbChannelMessageStore} should be used for those cases, and the last one with
* {@code priorityEnabled = true} option.
*
* @author Artem Bilan
* @since 4.0
*/
public class MongoDbChannelMessageStore extends AbstractConfigurableMongoDbMessageStore
implements PriorityCapableChannelMessageStore {
public final static String DEFAULT_COLLECTION_NAME = "channelMessages";
private volatile boolean priorityEnabled;
public MongoDbChannelMessageStore(MongoTemplate mongoTemplate) {
this(mongoTemplate, DEFAULT_COLLECTION_NAME);
}
public MongoDbChannelMessageStore(MongoTemplate mongoTemplate, String collectionName) {
super(mongoTemplate, collectionName);
}
public MongoDbChannelMessageStore(MongoDbFactory mongoDbFactory) {
this(mongoDbFactory, null, DEFAULT_COLLECTION_NAME);
}
public MongoDbChannelMessageStore(MongoDbFactory mongoDbFactory, MappingMongoConverter mappingMongoConverter) {
this(mongoDbFactory, mappingMongoConverter, DEFAULT_COLLECTION_NAME);
}
public MongoDbChannelMessageStore(MongoDbFactory mongoDbFactory, String collectionName) {
this(mongoDbFactory, null, collectionName);
}
public MongoDbChannelMessageStore(MongoDbFactory mongoDbFactory, MappingMongoConverter mappingMongoConverter, String collectionName) {
super(mongoDbFactory, mappingMongoConverter, collectionName);
}
public void setPriorityEnabled(boolean priorityEnabled) {
this.priorityEnabled = priorityEnabled;
}
@Override
public boolean isPriorityEnabled() {
return this.priorityEnabled;
}
@Override
public void afterPropertiesSet() throws Exception {
super.afterPropertiesSet();
this.mongoTemplate.indexOps(this.collectionName)
.ensureIndex(new Index(MessageDocumentFields.GROUP_ID, Order.ASCENDING)
.on(MessageDocumentFields.PRIORITY, Order.DESCENDING)
.on(MessageDocumentFields.LAST_MODIFIED_TIME, Order.ASCENDING)
.on(MessageDocumentFields.SEQUENCE, Order.ASCENDING));
}
@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 document = new MessageDocument(message);
document.setGroupId(groupId);
document.setCreatedTime(System.currentTimeMillis());
document.setLastModifiedTime(System.currentTimeMillis());
if (this.priorityEnabled) {
document.setPriority(new IntegrationMessageHeaderAccessor(message).getPriority());
}
document.setSequence(this.getNextId());
this.addMessageDocument(document);
return this.getMessageGroup(groupId);
}
/**
* Not fully used. Only wraps the provided group id.
*/
@Override
public MessageGroup getMessageGroup(Object groupId) {
return new SimpleMessageGroup(groupId);
}
@Override
public Message<?> pollMessageFromGroup(Object groupId) {
Assert.notNull(groupId, "'groupId' must not be null");
Sort sort = new Sort(MessageDocumentFields.LAST_MODIFIED_TIME, MessageDocumentFields.SEQUENCE);
if (this.priorityEnabled) {
sort = new Sort(Sort.Direction.DESC, MessageDocumentFields.PRIORITY).and(sort);
}
Query query = groupIdQuery(groupId).with(sort);
MessageDocument document = this.mongoTemplate.findAndRemove(query, MessageDocument.class, this.collectionName);
Message<?> message = null;
if (document != null) {
message = document.getMessage();
}
return message;
}
}

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

@@ -16,11 +16,6 @@
package org.springframework.integration.mongodb.store;
import static org.springframework.data.mongodb.core.query.Criteria.where;
import static org.springframework.integration.history.MessageHistory.NAME_PROPERTY;
import static org.springframework.integration.history.MessageHistory.TIMESTAMP_PROPERTY;
import static org.springframework.integration.history.MessageHistory.TYPE_PROPERTY;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
@@ -31,6 +26,12 @@ import java.util.Properties;
import java.util.Set;
import java.util.UUID;
import com.mongodb.BasicDBList;
import com.mongodb.BasicDBObject;
import com.mongodb.DB;
import com.mongodb.DBObject;
import com.mongodb.MongoException;
import org.springframework.beans.BeansException;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.BeanClassLoaderAware;
@@ -42,19 +43,25 @@ 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.dao.DataAccessException;
import org.springframework.data.annotation.Id;
import org.springframework.data.annotation.Transient;
import org.springframework.data.convert.WritingConverter;
import org.springframework.data.domain.Sort;
import org.springframework.data.domain.Sort.Direction;
import org.springframework.data.mapping.context.MappingContext;
import org.springframework.data.mongodb.MongoDbFactory;
import org.springframework.data.mongodb.core.DbCallback;
import org.springframework.data.mongodb.core.FindAndModifyOptions;
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.mapping.MongoPersistentEntity;
import org.springframework.data.mongodb.core.mapping.MongoPersistentProperty;
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.history.MessageHistory;
@@ -74,10 +81,6 @@ import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import org.springframework.util.StringUtils;
import com.mongodb.BasicDBList;
import com.mongodb.BasicDBObject;
import com.mongodb.DBObject;
/**
* An implementation of both the {@link MessageStore} and {@link MessageGroupStore}
@@ -96,6 +99,19 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore
private final static String DEFAULT_COLLECTION_NAME = "messages";
public final static String SEQUENCE_NAME = "messagesSequence";
/**
* 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 final static String GROUP_ID_KEY = "_groupId";
private final static String GROUP_COMPLETE_KEY = "_group_complete";
@@ -108,6 +124,8 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore
private final static String CREATED_DATE = "_createdDate";
private static final String SEQUENCE = "sequence";
private final MongoTemplate template;
@@ -161,15 +179,53 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore
this.converter.setApplicationContext(this.applicationContext);
}
this.converter.afterPropertiesSet();
IndexOperations indexOperations = this.template.indexOps(this.collectionName);
indexOperations.ensureIndex(new Index(GROUP_ID_KEY, Order.ASCENDING)
.on(GROUP_UPDATE_TIMESTAMP_KEY, Order.DESCENDING)
.on(SEQUENCE, Order.DESCENDING));
}
@Override
public <T> Message<T> addMessage(Message<T> message) {
Assert.notNull(message, "'message' must not be null");
this.template.insert(new MessageWrapper(message), this.collectionName);
this.addMessageDocument(new MessageWrapper(message));
return message;
}
private void addMessageDocument(final MessageWrapper document) {
this.template.executeInSession(new DbCallback<Void>() {
@Override
public Void doInDB(DB db) throws MongoException, DataAccessException {
Message<?> message = document.getMessage();
if (message.getHeaders().containsKey(SAVED_KEY)) {
Message<?> saved = getMessage(message.getHeaders().getId());
if (saved != null) {
if (saved.equals(message)) {
return null;
} // We need to save it under its own id
}
}
final long createdDate = document.get_Group_timestamp() == 0 ? System.currentTimeMillis() : document.get_Group_timestamp();
Message<?> result = getMessageBuilderFactory().fromMessage(message).setHeader(SAVED_KEY, Boolean.TRUE)
.setHeader(CREATED_DATE_KEY, createdDate).build();
@SuppressWarnings("unchecked")
Map<String, Object> innerMap = (Map<String, Object>) 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.set_Group_timestamp(createdDate);
template.insert(document, collectionName);
return null;
}
});
}
@Override
public Message<?> getMessage(UUID id) {
Assert.notNull(id, "'id' must not be null");
@@ -193,16 +249,17 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore
@Override
public MessageGroup getMessageGroup(Object groupId) {
Assert.notNull(groupId, "'groupId' must not be null");
List<MessageWrapper> messageWrappers = this.template.find(whereGroupIdIs(groupId), MessageWrapper.class, this.collectionName);
Query query = whereGroupIdOrder(groupId);
List<MessageWrapper> messageWrappers = this.template.find(query, MessageWrapper.class, this.collectionName);
List<Message<?>> messages = new ArrayList<Message<?>>();
long timestamp = 0;
long lastmodified = 0;
long lastModified = 0;
int lastReleasedSequenceNumber = 0;
boolean completeGroup = false;
if (messageWrappers.size() > 0){
MessageWrapper messageWrapper = messageWrappers.get(0);
timestamp = messageWrapper.get_Group_timestamp();
lastmodified = messageWrapper.get_Group_update_timestamp();
lastModified = messageWrapper.get_Group_update_timestamp();
completeGroup = messageWrapper.get_Group_complete();
lastReleasedSequenceNumber = messageWrapper.get_LastReleasedSequenceNumber();
}
@@ -212,7 +269,7 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore
}
SimpleMessageGroup messageGroup = new SimpleMessageGroup(messages, groupId, timestamp, completeGroup);
messageGroup.setLastModified(lastmodified);
messageGroup.setLastModified(lastModified);
if (lastReleasedSequenceNumber > 0){
messageGroup.setLastReleasedMessageSequenceNumber(lastReleasedSequenceNumber);
}
@@ -221,134 +278,178 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore
}
@Override
public MessageGroup addMessageToGroup(Object groupId, Message<?> message) {
public MessageGroup addMessageToGroup(final Object groupId, final Message<?> message) {
Assert.notNull(groupId, "'groupId' must not be null");
Assert.notNull(message, "'message' must not be null");
MessageGroup messageGroup = this.getMessageGroup(groupId);
return this.template.executeInSession(new DbCallback<MessageGroup>() {
long messageGroupTimestamp = messageGroup.getTimestamp();
long lastModified = messageGroup.getLastModified();
@Override
public MessageGroup doInDB(DB db) throws MongoException, DataAccessException {
Query query = whereGroupIdOrder(groupId);
MessageWrapper messageDocument = template.findOne(query, MessageWrapper.class, collectionName);
if (messageGroupTimestamp == 0){
messageGroupTimestamp = System.currentTimeMillis();
lastModified = messageGroupTimestamp;
}
else {
lastModified = System.currentTimeMillis();
}
long createdTime = 0;
int lastReleasedSequence = 0;
boolean complete = false;
MessageWrapper wrapper = new MessageWrapper(message);
wrapper.set_GroupId(groupId);
wrapper.set_Group_timestamp(messageGroupTimestamp);
wrapper.set_Group_update_timestamp(lastModified);
wrapper.set_Group_complete(messageGroup.isComplete());
wrapper.set_LastReleasedSequenceNumber(messageGroup.getLastReleasedMessageSequenceNumber());
if (messageDocument != null) {
createdTime = messageDocument.get_Group_timestamp();
lastReleasedSequence = messageDocument.get_LastReleasedSequenceNumber();
complete = messageDocument.get_Group_complete();
}
this.template.insert(wrapper, this.collectionName);
return this.getMessageGroup(groupId);
MessageWrapper wrapper = new MessageWrapper(message);
wrapper.set_GroupId(groupId);
wrapper.set_Group_timestamp(createdTime == 0 ? System.currentTimeMillis() : createdTime);
wrapper.set_Group_update_timestamp(System.currentTimeMillis());
wrapper.set_Group_complete(complete);
wrapper.set_LastReleasedSequenceNumber(lastReleasedSequence);
wrapper.setSequence(getNextId());
addMessageDocument(wrapper);
return getMessageGroup(groupId);
}
});
}
@Override
public MessageGroup removeMessageFromGroup(Object groupId, Message<?> messageToRemove) {
public MessageGroup removeMessageFromGroup(final Object groupId, final Message<?> messageToRemove) {
Assert.notNull(groupId, "'groupId' must not be null");
Assert.notNull(messageToRemove, "'messageToRemove' must not be null");
this.template.findAndRemove(whereMessageIdIsAndGroupIdIs(
messageToRemove.getHeaders().getId(), groupId), MessageWrapper.class, this.collectionName);
this.updateGroup(groupId);
return this.getMessageGroup(groupId);
return this.template.executeInSession(new DbCallback<MessageGroup>() {
@Override
public MessageGroup doInDB(DB db) throws MongoException, DataAccessException {
template.findAndRemove(whereMessageIdIsAndGroupIdIs(messageToRemove.getHeaders().getId(), groupId),
MessageWrapper.class, collectionName);
updateGroup(groupId, lastModifiedUpdate());
return getMessageGroup(groupId);
}
});
}
@Override
public void removeMessageGroup(Object groupId) {
List<MessageWrapper> messageWrappers = this.template.find(whereGroupIdIs(groupId), MessageWrapper.class, this.collectionName);
for (MessageWrapper messageWrapper : messageWrappers) {
this.removeMessageFromGroup(groupId, messageWrapper.getMessage());
}
this.template.remove(whereGroupIdIs(groupId), this.collectionName);
}
@Override
public Iterator<MessageGroup> iterator() {
List<MessageWrapper> groupedMessages = this.template.find(whereGroupIdExists(), MessageWrapper.class, this.collectionName);
Map<Object, MessageGroup> messageGroups = new HashMap<Object, MessageGroup>();
for (MessageWrapper groupedMessage : groupedMessages) {
Object groupId = groupedMessage.get_GroupId();
if (!messageGroups.containsKey(groupId)) {
messageGroups.put(groupId, this.getMessageGroup(groupId));
return this.template.executeInSession(new DbCallback<Iterator<MessageGroup>>() {
@Override
public Iterator<MessageGroup> doInDB(DB db) throws MongoException, DataAccessException {
List<MessageGroup> messageGroups = new ArrayList<MessageGroup>();
Query query = Query.query(Criteria.where(GROUP_ID_KEY).exists(true));
@SuppressWarnings("rawtypes")
List groupIds = template.getCollection(collectionName)
.distinct(GROUP_ID_KEY, query.getQueryObject());
for (Object groupId : groupIds) {
messageGroups.add(getMessageGroup(groupId));
}
return messageGroups.iterator();
}
}
return messageGroups.values().iterator();
});
}
@Override
public void completeGroup(Object groupId) {
Update update = Update.update(GROUP_COMPLETE_KEY, true);
Query q = whereGroupIdIs(groupId);
this.template.updateFirst(q, update, this.collectionName);
this.updateGroup(groupId);
}
@Override
public void setLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber) {
Update update = Update.update(LAST_RELEASED_SEQUENCE_NUMBER, sequenceNumber);
Query q = whereGroupIdIs(groupId);
this.template.updateFirst(q, update, this.collectionName);
this.updateGroup(groupId);
}
@Override
public Message<?> pollMessageFromGroup(Object groupId) {
public Message<?> pollMessageFromGroup(final Object groupId) {
Assert.notNull(groupId, "'groupId' must not be null");
MessageWrapper messageWrapper = this.template.findAndRemove(whereGroupIdIsOrdered(groupId), MessageWrapper.class, this.collectionName);
Message<?> message = null;
if (messageWrapper != null) {
message = messageWrapper.getMessage();
}
this.updateGroup(groupId);
return message;
return this.template.executeInSession(new DbCallback<Message<?>>() {
@Override
public Message<?> doInDB(DB db) throws MongoException, DataAccessException {
Query query = whereGroupIdIs(groupId).with(new Sort(GROUP_UPDATE_TIMESTAMP_KEY, SEQUENCE));
MessageWrapper messageWrapper = template.findAndRemove(query, MessageWrapper.class, collectionName);
Message<?> message = null;
if (messageWrapper != null) {
message = messageWrapper.getMessage();
}
updateGroup(groupId, lastModifiedUpdate());
return message;
}
});
}
@Override
public int messageGroupSize(Object groupId) {
long lCount = this.template.count(new Query(where(GROUP_ID_KEY).is(groupId)), this.collectionName);
long lCount = this.template.count(new Query(Criteria.where(GROUP_ID_KEY).is(groupId)), this.collectionName);
Assert.isTrue(lCount <= Integer.MAX_VALUE, "Message count is out of Integer's range");
return (int) lCount;
}
@Override
public void setLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber) {
this.updateGroup(groupId, lastModifiedUpdate().set(LAST_RELEASED_SEQUENCE_NUMBER, sequenceNumber));
}
@Override
public void completeGroup(Object groupId) {
this.updateGroup(groupId, lastModifiedUpdate().set(GROUP_COMPLETE_KEY, true));
}
@Override
@ManagedAttribute
public int getMessageCountForAllMessageGroups() {
Query query = Query.query(Criteria.where(MessageDocumentFields.MESSAGE_ID).exists(true)
.and(MessageDocumentFields.GROUP_ID).exists(true));
long count = this.template.count(query, this.collectionName);
Assert.isTrue(count <= Integer.MAX_VALUE, "Message count is out of Integer's range");
return (int) count;
}
@Override
@ManagedAttribute
public int getMessageGroupCount() {
Query query = Query.query(Criteria.where(MessageDocumentFields.GROUP_ID).exists(true));
return this.template.getCollection(this.collectionName)
.distinct(MessageDocumentFields.GROUP_ID, query.getQueryObject())
.size();
}
private static Update lastModifiedUpdate() {
return Update.update(GROUP_UPDATE_TIMESTAMP_KEY, System.currentTimeMillis());
}
/*
* Common Queries
*/
private static Query whereMessageIdIs(UUID id) {
return new Query(where("headers.id._value").is(id.toString()));
return new Query(Criteria.where("headers.id._value").is(id.toString()));
}
private static Query whereMessageIdIsAndGroupIdIs(UUID id, Object groupId) {
return new Query(where("headers.id._value").is(id.toString()).and(GROUP_ID_KEY).is(groupId));
return new Query(Criteria.where("headers.id._value").is(id.toString()).and(GROUP_ID_KEY).is(groupId));
}
private static Query whereGroupIdOrder(Object groupId) {
return whereGroupIdIs(groupId).with(new Sort(Sort.Direction.DESC, GROUP_UPDATE_TIMESTAMP_KEY, SEQUENCE));
}
private static Query whereGroupIdIs(Object groupId) {
Query q = new Query(where(GROUP_ID_KEY).is(groupId));
q.with(new Sort(Direction.DESC, GROUP_UPDATE_TIMESTAMP_KEY));
return q;
return new Query(Criteria.where(GROUP_ID_KEY).is(groupId));
}
private static Query whereGroupIdExists() {
return new Query(where(GROUP_ID_KEY).exists(true));
private void updateGroup(Object groupId, Update update) {
Query query = whereGroupIdIs(groupId).with(new Sort(Sort.Direction.DESC, GROUP_UPDATE_TIMESTAMP_KEY, SEQUENCE));
this.template.updateFirst(query, update, this.collectionName);
}
private static Query whereGroupIdIsOrdered(Object groupId) {
Query q = new Query(where(GROUP_ID_KEY).is(groupId)).limit(1);
q.with(new Sort(Direction.ASC, CREATED_DATE));
return q;
private int getNextId() {
Query query = Query.query(Criteria.where("_id").is(SEQUENCE_NAME));
query.fields().include(SEQUENCE);
return (Integer) this.template.findAndModify(query,
new Update().inc(SEQUENCE, 1),
FindAndModifyOptions.options().returnNew(true).upsert(true),
Map.class,
this.collectionName).get(SEQUENCE);
}
private void updateGroup(Object groupId) {
Update update = Update.update(GROUP_UPDATE_TIMESTAMP_KEY, System.currentTimeMillis());
Query q = whereGroupIdIs(groupId);
this.template.updateFirst(q, update, this.collectionName);
}
/**
* Custom implementation of the {@link MappingMongoConverter} strategy.
*/
@@ -516,9 +617,9 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore
BasicDBList dbList = new BasicDBList();
for (Properties properties : source) {
BasicDBObject dbo = new BasicDBObject();
dbo.put(NAME_PROPERTY, properties.getProperty(NAME_PROPERTY));
dbo.put(TYPE_PROPERTY, properties.getProperty(TYPE_PROPERTY));
dbo.put(TIMESTAMP_PROPERTY, properties.getProperty(TIMESTAMP_PROPERTY));
dbo.put(MessageHistory.NAME_PROPERTY, properties.getProperty(MessageHistory.NAME_PROPERTY));
dbo.put(MessageHistory.TYPE_PROPERTY, properties.getProperty(MessageHistory.TYPE_PROPERTY));
dbo.put(MessageHistory.TIMESTAMP_PROPERTY, properties.getProperty(MessageHistory.TIMESTAMP_PROPERTY));
dbList.add(dbo);
}
obj.put("components", dbList);
@@ -654,11 +755,13 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore
@SuppressWarnings("unused")
private final String _messageType;
@SuppressWarnings("unused")
private final Object payload;
@SuppressWarnings("unused")
private final Map<String, ?> headers;
@SuppressWarnings("unused")
private final Message<?> inputMessage;
private volatile long _group_timestamp;
@@ -669,6 +772,9 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore
private volatile boolean _group_complete;
@SuppressWarnings("unused")
private int sequence;
public MessageWrapper(Message<?> message) {
Assert.notNull(message, "'message' must not be null");
this.message = message;
@@ -726,5 +832,11 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore
public void set_Group_complete(boolean completedGroup) {
this._group_complete = completedGroup;
}
public void setSequence(int sequence) {
this.sequence = sequence;
}
}
}

9
spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/AbstractMongoDbMessageGroupStoreTests.java
View File

@@ -15,17 +15,14 @@
*/
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 static org.junit.Assert.*;
import java.util.Collection;
import java.util.Iterator;
import java.util.Properties;
import java.util.UUID;
import com.mongodb.Mongo;
import org.junit.Test;
import org.springframework.context.support.ClassPathXmlApplicationContext;
@@ -44,8 +41,6 @@ import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.GenericMessage;
import com.mongodb.Mongo;
/**
* @author Oleg Zhurakousky
* @author Gary Russell

75
spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/ConfigurableMongoDbMessageGroupStoreTests.java
View File

@@ -15,10 +15,13 @@
*/
package org.springframework.integration.mongodb.store;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.*;
import java.util.Map;
import com.mongodb.DBObject;
import com.mongodb.Mongo;
import org.hamcrest.Matchers;
import org.junit.Test;
import org.springframework.context.support.ClassPathXmlApplicationContext;
@@ -26,15 +29,15 @@ 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.IntegrationMessageHeaderAccessor;
import org.springframework.integration.channel.PriorityChannel;
import org.springframework.integration.channel.QueueChannel;
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 org.springframework.messaging.Message;
import com.mongodb.DBObject;
import com.mongodb.Mongo;
/**
* @author Amol Nayak
* @author Artem Bilan
@@ -73,7 +76,7 @@ public class ConfigurableMongoDbMessageGroupStoreTests extends AbstractMongoDbMe
@Test
@MongoDbAvailable
public void testWithCustomConverter() throws Exception {
this.prepareMongoFactory("testConfigurableMongoDbMessageStore");
this.cleanupCollections(new SimpleMongoDbFactory(new Mongo(), "test"));
ClassPathXmlApplicationContext context =
new ClassPathXmlApplicationContext("ConfigurableMongoDbMessageStore-CustomConverter.xml", this.getClass());
context.refresh();
@@ -84,6 +87,68 @@ public class ConfigurableMongoDbMessageGroupStoreTests extends AbstractMongoDbMe
context.close();
}
@Test
@MongoDbAvailable
public void testPriorityChannel() throws Exception {
this.cleanupCollections(new SimpleMongoDbFactory(new Mongo(), "test"));
ClassPathXmlApplicationContext context =
new ClassPathXmlApplicationContext("ConfigurableMongoDbMessageStore-CustomConverter.xml", this.getClass());
context.refresh();
Object priorityChannel = context.getBean("priorityChannel");
assertThat(priorityChannel, Matchers.not(Matchers.instanceOf(PriorityChannel.class)));
assertThat(priorityChannel, Matchers.instanceOf(QueueChannel.class));
QueueChannel channel = (QueueChannel) priorityChannel;
Message<String> message = MessageBuilder.withPayload("1").setHeader(IntegrationMessageHeaderAccessor.PRIORITY, 1).build();
channel.send(message);
message = MessageBuilder.withPayload("-1").setHeader(IntegrationMessageHeaderAccessor.PRIORITY, -1).build();
channel.send(message);
message = MessageBuilder.withPayload("3").setHeader(IntegrationMessageHeaderAccessor.PRIORITY, 3).build();
channel.send(message);
message = MessageBuilder.withPayload("0").setHeader(IntegrationMessageHeaderAccessor.PRIORITY, 0).build();
channel.send(message);
message = MessageBuilder.withPayload("2").setHeader(IntegrationMessageHeaderAccessor.PRIORITY, 2).build();
channel.send(message);
message = MessageBuilder.withPayload("none").build();
channel.send(message);
message = MessageBuilder.withPayload("31").setHeader(IntegrationMessageHeaderAccessor.PRIORITY, 3).build();
channel.send(message);
Message<?> receive = channel.receive(1000);
assertNotNull(receive);
assertEquals("3", receive.getPayload());
receive = channel.receive(1000);
assertNotNull(receive);
assertEquals("31", receive.getPayload());
receive = channel.receive(1000);
assertNotNull(receive);
assertEquals("2", receive.getPayload());
receive = channel.receive(1000);
assertNotNull(receive);
assertEquals("1", receive.getPayload());
receive = channel.receive(1000);
assertNotNull(receive);
assertEquals("0", receive.getPayload());
receive = channel.receive(1000);
assertNotNull(receive);
assertEquals("-1", receive.getPayload());
receive = channel.receive(1000);
assertNotNull(receive);
assertEquals("none", receive.getPayload());
context.close();
}
public static interface TestGateway {
String service(String payload);

37
spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/ConfigurableMongoDbMessageStore-CustomConverter.xml
View File

@@ -12,18 +12,21 @@
<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 id="abstractMessageStore" class="org.springframework.integration.mongodb.store.MongoDbChannelMessageStore"
abstract="true">
<beans:constructor-arg name="mongoDbFactory" ref="mongoDbFactory"/>
</beans:bean>
<mongo:mapping-converter id="customConverter">
<mongo:custom-converters>
<mongo:converter>
<beans:bean class="org.springframework.integration.mongodb.store.ConfigurableMongoDbMessageGroupStoreTests$MessageReadConverter"/>
</mongo:converter>
</mongo:custom-converters>
</mongo:mapping-converter>
<beans:bean id="messageStore" parent="abstractMessageStore">
<beans:constructor-arg name="mappingMongoConverter" ref="customConverter"/>
</beans:bean>
<gateway id="gateway"
@@ -42,4 +45,14 @@
<poller fixed-delay="1000"/>
</transformer>
<poller default="true" fixed-delay="10000"/>
<beans:bean id="priorityMessageStore" parent="abstractMessageStore">
<beans:property name="priorityEnabled" value="true"/>
</beans:bean>
<channel id="priorityChannel">
<priority-queue message-store="priorityMessageStore"/>
</channel>
</beans:beans>

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

@@ -15,14 +15,13 @@
*/
package org.springframework.integration.mongodb.store;
import com.mongodb.Mongo;
import org.junit.Test;
import org.springframework.data.mongodb.core.SimpleMongoDbFactory;
import org.springframework.integration.mongodb.rules.MongoDbAvailable;
import org.springframework.integration.store.MessageStore;
import com.mongodb.Mongo;
/**
* @author Oleg Zhurakousky
* @author Gary Russell
@@ -48,4 +47,5 @@ public class MongoDbMessageGroupStoreTests extends AbstractMongoDbMessageGroupSt
public void testWithAggregatorWithShutdown() throws Exception {
super.testWithAggregatorWithShutdown("mongo-aggregator-config.xml");
}
}