diff --git a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/AbstractConfigurableMongoDbMessageStore.java b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/AbstractConfigurableMongoDbMessageStore.java index cfc917e0a5..349bdc0182 100644 --- a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/AbstractConfigurableMongoDbMessageStore.java +++ b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/AbstractConfigurableMongoDbMessageStore.java @@ -150,13 +150,15 @@ public abstract class AbstractConfigurableMongoDbMessageStore extends AbstractMe indexOperations.ensureIndex(new Index(MessageDocumentFields.MESSAGE_ID, Sort.Direction.ASC)); - indexOperations.ensureIndex(new Index(MessageDocumentFields.GROUP_ID, Sort.Direction.ASC) - .on(MessageDocumentFields.MESSAGE_ID, Sort.Direction.ASC) - .unique()); + indexOperations.ensureIndex( + new Index(MessageDocumentFields.GROUP_ID, Sort.Direction.ASC) + .on(MessageDocumentFields.MESSAGE_ID, Sort.Direction.ASC) + .unique()); - indexOperations.ensureIndex(new Index(MessageDocumentFields.GROUP_ID, Sort.Direction.ASC) - .on(MessageDocumentFields.LAST_MODIFIED_TIME, Sort.Direction.DESC) - .on(MessageDocumentFields.SEQUENCE, Sort.Direction.DESC)); + indexOperations.ensureIndex( + new Index(MessageDocumentFields.GROUP_ID, Sort.Direction.ASC) + .on(MessageDocumentFields.LAST_MODIFIED_TIME, Sort.Direction.DESC) + .on(MessageDocumentFields.SEQUENCE, Sort.Direction.DESC)); } public Message getMessage(UUID id) { @@ -198,14 +200,15 @@ public abstract class AbstractConfigurableMongoDbMessageStore extends AbstractMe * The {@link #SEQUENCE_NAME} document is created on demand. * @return the next sequence value. */ - protected int getNextId() { + protected long 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), + return ((Number) this.mongoTemplate.findAndModify(query, + new Update().inc(MessageDocumentFields.SEQUENCE, 1L), FindAndModifyOptions.options().returnNew(true).upsert(true), Map.class, this.collectionName) - .get(MessageDocumentFields.SEQUENCE); // NOSONAR - never returns null + .get(MessageDocumentFields.SEQUENCE)) // NOSONAR - never returns null + .longValue(); } protected void addMessageDocument(final MessageDocument document) { diff --git a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MessageDocument.java b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MessageDocument.java index 3f6fc9b3da..0507ba4cd5 100644 --- a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MessageDocument.java +++ b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MessageDocument.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2019 the original author or authors. + * Copyright 2014-2020 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. @@ -61,7 +61,7 @@ public class MessageDocument { private Integer lastReleasedSequence = 0; - private int sequence; + private long sequence; public MessageDocument(Message message) { this(message, message.getHeaders().getId()); @@ -139,7 +139,7 @@ public class MessageDocument { this.lastReleasedSequence = lastReleasedSequence; } - public void setSequence(int sequence) { + public void setSequence(long sequence) { this.sequence = sequence; } @@ -151,7 +151,7 @@ public class MessageDocument { return this.groupId; } - public int getSequence() { + public long getSequence() { return this.sequence; } diff --git a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoDbChannelMessageStore.java b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoDbChannelMessageStore.java index 17173ddc2f..4e90dcf733 100644 --- a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoDbChannelMessageStore.java +++ b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoDbChannelMessageStore.java @@ -49,6 +49,9 @@ import org.springframework.util.Assert; public class MongoDbChannelMessageStore extends AbstractConfigurableMongoDbMessageStore implements PriorityCapableChannelMessageStore { + /** + * The default conventional collection name. + */ public static final String DEFAULT_COLLECTION_NAME = "channelMessages"; private boolean priorityEnabled; @@ -65,7 +68,9 @@ public class MongoDbChannelMessageStore extends AbstractConfigurableMongoDbMessa this(mongoDbFactory, null, DEFAULT_COLLECTION_NAME); } - public MongoDbChannelMessageStore(MongoDatabaseFactory mongoDbFactory, MappingMongoConverter mappingMongoConverter) { + public MongoDbChannelMessageStore(MongoDatabaseFactory mongoDbFactory, + MappingMongoConverter mappingMongoConverter) { + this(mongoDbFactory, mappingMongoConverter, DEFAULT_COLLECTION_NAME); } @@ -75,6 +80,7 @@ public class MongoDbChannelMessageStore extends AbstractConfigurableMongoDbMessa public MongoDbChannelMessageStore(MongoDatabaseFactory mongoDbFactory, MappingMongoConverter mappingMongoConverter, String collectionName) { + super(mongoDbFactory, mappingMongoConverter, collectionName); } @@ -90,7 +96,8 @@ public class MongoDbChannelMessageStore extends AbstractConfigurableMongoDbMessa @Override public void afterPropertiesSet() { super.afterPropertiesSet(); - getMongoTemplate().indexOps(this.collectionName) + getMongoTemplate() + .indexOps(this.collectionName) .ensureIndex(new Index(MessageDocumentFields.GROUP_ID, Sort.Direction.ASC) .on(MessageDocumentFields.PRIORITY, Sort.Direction.DESC) .on(MessageDocumentFields.LAST_MODIFIED_TIME, Sort.Direction.ASC) @@ -109,10 +116,10 @@ public class MongoDbChannelMessageStore extends AbstractConfigurableMongoDbMessa if (this.priorityEnabled) { document.setPriority(message.getHeaders().get(IntegrationMessageHeaderAccessor.PRIORITY, Integer.class)); } - document.setSequence(this.getNextId()); + document.setSequence(getNextId()); - this.addMessageDocument(document); - return this.getMessageGroup(groupId); + addMessageDocument(document); + return getMessageGroup(groupId); } /** diff --git a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoDbMessageStore.java b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoDbMessageStore.java index f73c168a06..2a9277a206 100644 --- a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoDbMessageStore.java +++ b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoDbMessageStore.java @@ -470,14 +470,15 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore this.template.findAndModify(query, update, FindAndModifyOptions.none(), Map.class, this.collectionName); } - private int getNextId() { + private long 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), + return ((Number) this.template.findAndModify(query, + new Update().inc(SEQUENCE, 1L), FindAndModifyOptions.options().returnNew(true).upsert(true), - Map.class, - this.collectionName).get(SEQUENCE); // NOSONAR - never returns null + Map.class, this.collectionName) + .get(SEQUENCE)) // NOSONAR - never returns null + .longValue(); } @SuppressWarnings(UNCHECKED) @@ -848,7 +849,7 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore private volatile boolean _group_complete; // NOSONAR name @SuppressWarnings(UNUSED) - private int sequence; + private long sequence; MessageWrapper(Message message) { Assert.notNull(message, "'message' must not be null"); @@ -917,7 +918,7 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore this._group_complete = completedGroup; } - public void set_Sequence(int sequence) { // NOSONAR name + public void set_Sequence(long sequence) { // NOSONAR name this.sequence = sequence; }