From 50d8caa1c33221c96e247c51072b60bb4109e820 Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Mon, 21 Nov 2011 04:54:13 -0500 Subject: [PATCH] INT-2231 Fixed JdbcMessageStore to reuse CREATED_DATE while updating UPDATED_DATE Added support for configuring MessageGroup timeout based on the idle time of the MessageGroup Changed updatedTimestamp to lastModified Added an assertion to AbsractMessageGroup.expireMessageGroups to check if lastModified is 0 Changed attribute name in MessageWrapper in Mongo Message Store --- .../store/AbstractKeyValueMessageStore.java | 5 +- .../store/AbstractMessageGroupStore.java | 24 ++++- .../integration/store/MessageGroup.java | 5 + .../store/MessageGroupMetadata.java | 7 ++ .../integration/store/SimpleMessageGroup.java | 10 ++ .../integration/jdbc/JdbcMessageStore.java | 97 +++++++++---------- .../jdbc/JdbcMessageStoreTests.java | 60 +++++++++--- .../mongodb/store/MongoDbMessageStore.java | 41 +++++++- .../store/MongoDbMessageGroupStoreTests.java | 29 ++++++ .../store/RedisMessageGroupStoreTests.java | 27 ++++++ 10 files changed, 237 insertions(+), 68 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractKeyValueMessageStore.java b/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractKeyValueMessageStore.java index 3f780204cd..a41a94931f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractKeyValueMessageStore.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractKeyValueMessageStore.java @@ -109,14 +109,14 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS // build raw MessageGroup and add enriched Message to it SimpleMessageGroup rawGroup = this.buildMessageGroup(groupId, true); - + rawGroup.setLastModified(System.currentTimeMillis()); rawGroup.add(enrichedMessage); // store MessageGroupMetadata built from enriched MG this.doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, new MessageGroupMetadata(rawGroup)); // return clean MG - return messageGroup; + return this.getMessageGroup(groupId); } /** @@ -250,6 +250,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS SimpleMessageGroup messageGroup = new SimpleMessageGroup(messages, groupId, messageGroupMetadata.getTimestamp(), messageGroupMetadata.isComplete()); + messageGroup.setLastModified(messageGroupMetadata.getLastModified()); messageGroup.setLastReleasedMessageSequenceNumber(messageGroupMetadata.getLastReleasedMessageSequenceNumber()); return messageGroup; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractMessageGroupStore.java b/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractMessageGroupStore.java index c55632ffd8..a8f106f846 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractMessageGroupStore.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractMessageGroupStore.java @@ -32,6 +32,8 @@ public abstract class AbstractMessageGroupStore implements MessageGroupStore, It protected final Log logger = LogFactory.getLog(getClass()); private Collection expiryCallbacks = new LinkedHashSet(); + + private volatile boolean timeoutOnIdle; /** * @@ -51,6 +53,20 @@ public abstract class AbstractMessageGroupStore implements MessageGroupStore, It registerMessageGroupExpiryCallback(callback); } } + + public boolean isTimeoutOnIdle() { + return timeoutOnIdle; + } + + /** + * 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'. + */ + public void setTimeoutOnIdle(boolean timeoutOnIdle) { + this.timeoutOnIdle = timeoutOnIdle; + } public void registerMessageGroupExpiryCallback(MessageGroupCallback callback) { expiryCallbacks.add(callback); @@ -60,7 +76,13 @@ public abstract class AbstractMessageGroupStore implements MessageGroupStore, It int count = 0; long threshold = System.currentTimeMillis() - timeout; for (MessageGroup group : this) { - if (group.getTimestamp() <= threshold) { + + long timestamp = group.getTimestamp(); + if (this.isTimeoutOnIdle() && group.getLastModified() > 0) { + timestamp = group.getLastModified(); + } + + if (timestamp <= threshold) { count++; expire(group); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroup.java b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroup.java index b479536cc8..b5a9960c88 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroup.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroup.java @@ -80,5 +80,10 @@ public interface MessageGroup { * @return the timestamp (milliseconds since epoch) associated with the creation of this group */ long getTimestamp(); + + /** + * @return the timestamp (milliseconds since epoch) associated with the time this group was last updated + */ + long getLastModified(); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupMetadata.java b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupMetadata.java index 91d099fe56..d0dd87ff4b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupMetadata.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupMetadata.java @@ -44,6 +44,8 @@ public class MessageGroupMetadata implements Serializable{ private final boolean complete; private final long timestamp; + + private volatile long lastModified; private final int lastReleasedMessageSequenceNumber; @@ -61,6 +63,7 @@ public class MessageGroupMetadata implements Serializable{ this.complete = messageGroup.isComplete(); this.timestamp = messageGroup.getTimestamp(); this.lastReleasedMessageSequenceNumber = messageGroup.getLastReleasedMessageSequenceNumber(); + this.lastModified = messageGroup.getLastModified(); } public void remove(UUID messageId){ @@ -93,6 +96,10 @@ public class MessageGroupMetadata implements Serializable{ public boolean isComplete() { return this.complete; } + + public long getLastModified() { + return lastModified; + } public long getTimestamp() { return this.timestamp; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageGroup.java b/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageGroup.java index 92fbe3eb83..03c105caf8 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageGroup.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageGroup.java @@ -39,6 +39,8 @@ public class SimpleMessageGroup implements MessageGroup { private final long timestamp; + private volatile long lastModified; + private volatile boolean complete; public SimpleMessageGroup(Object groupId) { @@ -65,6 +67,14 @@ public class SimpleMessageGroup implements MessageGroup { public long getTimestamp() { return timestamp; } + + public void setLastModified(long lastModified){ + this.lastModified = lastModified; + } + + public long getLastModified() { + return lastModified; + } public boolean canAdd(Message message) { return !isMember(message); diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java index cd6afe18e0..d67cfa185b 100644 --- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java @@ -79,6 +79,8 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa public static final String DEFAULT_TABLE_PREFIX = "INT_"; private static final String GET_MESSAGE = "SELECT MESSAGE_ID, CREATED_DATE, MESSAGE_BYTES from %PREFIX%MESSAGE where MESSAGE_ID=? and REGION=?"; + + private static final String GET_GROUP_CREATED_DATE = "SELECT CREATED_DATE from %PREFIX%MESSAGE_GROUP where GROUP_KEY=? and REGION=?"; private static final String GET_MESSAGE_COUNT = "SELECT COUNT(MESSAGE_ID) from %PREFIX%MESSAGE where REGION=?"; @@ -87,9 +89,9 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa private static final String CREATE_MESSAGE = "INSERT into %PREFIX%MESSAGE(MESSAGE_ID, REGION, CREATED_DATE, MESSAGE_BYTES)" + " values (?, ?, ?, ?)"; - private static final String LIST_MESSAGES_BY_GROUP_KEY = "SELECT MESSAGE_ID, CREATED_DATE, GROUP_KEY, MESSAGE_BYTES, MARKED, COMPLETE, LAST_RELEASED_SEQUENCE from %PREFIX%MESSAGE_GROUP where GROUP_KEY=? and REGION=? order by CREATED_DATE"; + private static final String LIST_MESSAGES_BY_GROUP_KEY = "SELECT MESSAGE_ID, CREATED_DATE, UPDATED_DATE, GROUP_KEY, MESSAGE_BYTES, MARKED, COMPLETE, LAST_RELEASED_SEQUENCE from %PREFIX%MESSAGE_GROUP where GROUP_KEY=? and REGION=? order by UPDATED_DATE"; - private static final String LIST_MESSAGEIDS_BY_GROUP_KEY = "SELECT MESSAGE_ID, CREATED_DATE from %PREFIX%MESSAGE_GROUP where GROUP_KEY=? and REGION=? order by CREATED_DATE"; + private static final String LIST_MESSAGEIDS_BY_GROUP_KEY = "SELECT MESSAGE_ID, CREATED_DATE from %PREFIX%MESSAGE_GROUP where GROUP_KEY=? and REGION=? order by UPDATED_DATE"; private static final String COUNT_ALL_GROUPS = "SELECT COUNT(GROUP_KEY) from %PREFIX%MESSAGE_GROUP where REGION=?"; @@ -97,10 +99,6 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa private static final String COUNT_ALL_MESSAGES_IN_GROUPS = "SELECT COUNT(MESSAGE_ID) from %PREFIX%MESSAGE_GROUP where REGION=?"; - private static final String MARK_MESSAGES_IN_GROUP = "UPDATE %PREFIX%MESSAGE_GROUP set UPDATED_DATE=?, MARKED=1 where MARKED=0 and GROUP_KEY=? and REGION=?"; - - private static final String MARK_MESSAGE_IN_GROUP = "UPDATE %PREFIX%MESSAGE_GROUP set UPDATED_DATE=?, MARKED=1 where MESSAGE_ID=? and MARKED=0 and GROUP_KEY=? and REGION=?"; - private static final String COMPLETE_GROUP = "UPDATE %PREFIX%MESSAGE_GROUP set UPDATED_DATE=?, COMPLETE=1 where GROUP_KEY=? and REGION=?"; private static final String UPDATE_LAST_RELEASED_SEQUENCE = "UPDATE %PREFIX%MESSAGE_GROUP set UPDATED_DATE=?, LAST_RELEASED_SEQUENCE=? where GROUP_KEY=? and REGION=?"; @@ -109,8 +107,8 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa private static final String DELETE_MESSAGE_GROUP = "DELETE from %PREFIX%MESSAGE_GROUP where GROUP_KEY=? and REGION=?"; - private static final String CREATE_MESSAGE_IN_GROUP = "INSERT into %PREFIX%MESSAGE_GROUP(MESSAGE_ID, REGION, CREATED_DATE, GROUP_KEY, MARKED, COMPLETE, LAST_RELEASED_SEQUENCE)" - + " values (?, ?, ?, ?, 0, 0, 0)"; + private static final String CREATE_MESSAGE_IN_GROUP = "INSERT into %PREFIX%MESSAGE_GROUP(MESSAGE_ID, REGION, CREATED_DATE, UPDATED_DATE, GROUP_KEY, MARKED, COMPLETE, LAST_RELEASED_SEQUENCE)" + + " values (?, ?, ?, ?, ?, 0, 0, 0)"; private static final String LIST_GROUP_KEYS = "SELECT distinct GROUP_KEY as CREATED from %PREFIX%MESSAGE_GROUP where REGION=?"; @@ -309,18 +307,27 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa } public MessageGroup addMessageToGroup(Object groupId, Message message) { - - final long createdDate = System.currentTimeMillis(); - final String messageId = getKey(message.getHeaders().getId()); final String groupKey = getKey(groupId); + final long updatedDate = System.currentTimeMillis(); + final long createdDate = this.getGroupCreatedDate(groupKey); + + final String messageId = getKey(message.getHeaders().getId()); + jdbcTemplate.update(getQuery(CREATE_MESSAGE_IN_GROUP), new PreparedStatementSetter() { public void setValues(PreparedStatement ps) throws SQLException { logger.debug("Inserting message with id key=" + messageId + " and created date=" + createdDate); ps.setString(1, messageId); ps.setString(2, region); - ps.setTimestamp(3, new Timestamp(createdDate)); - ps.setString(4, groupKey); + if (createdDate == 0){ + ps.setTimestamp(3, new Timestamp(updatedDate)); + } + else { + ps.setTimestamp(3, new Timestamp(createdDate)); + } + + ps.setTimestamp(4, new Timestamp(updatedDate)); + ps.setString(5, groupKey); } }); this.addMessage(message); @@ -347,6 +354,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa String key = getKey(groupId); final List> messages = new ArrayList>(); final AtomicReference date = new AtomicReference(); + final AtomicReference updateDate = new AtomicReference(); final AtomicReference completeFlag = new AtomicReference(); final AtomicReference lastReleasedSequenceRef = new AtomicReference(); @@ -361,6 +369,8 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa date.set(rs.getTimestamp("CREATED_DATE")); + updateDate.set(rs.getTimestamp("UPDATED_DATE")); + completeFlag.set(rs.getInt("COMPLETE") > 0); lastReleasedSequenceRef.set(rs.getInt("LAST_RELEASED_SEQUENCE")); @@ -371,9 +381,13 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa return new SimpleMessageGroup(groupId); } Assert.state(date.get() != null, "Could not locate created date for groupId=" + groupId); + Assert.state(updateDate.get() != null, "Could not locate updated date for groupId=" + groupId); long timestamp = date.get().getTime(); boolean complete = completeFlag.get().booleanValue(); SimpleMessageGroup messageGroup = new SimpleMessageGroup(messages, groupId, timestamp, complete); + if (updateDate.get() != null){ + messageGroup.setLastModified(updateDate.get().getTime()); + } int lastReleasedSequenceNumber = lastReleasedSequenceRef.get(); if (lastReleasedSequenceNumber > 0){ messageGroup.setLastReleasedMessageSequenceNumber(lastReleasedSequenceNumber); @@ -382,24 +396,6 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa return messageGroup; } - public MessageGroup markMessageGroup(MessageGroup group) { - - final long updatedDate = System.currentTimeMillis(); - final String groupKey = getKey(group.getGroupId()); - - jdbcTemplate.update(getQuery(MARK_MESSAGES_IN_GROUP), new PreparedStatementSetter() { - public void setValues(PreparedStatement ps) throws SQLException { - logger.debug("Marking messages with group key=" + groupKey); - ps.setTimestamp(1, new Timestamp(updatedDate)); - ps.setString(2, groupKey); - ps.setString(3, region); - } - }); - - return getMessageGroup(group.getGroupId()); - - } - public MessageGroup removeMessageFromGroup(Object groupId, Message messageToRemove) { final String groupKey = getKey(groupId); final String messageId = getKey(messageToRemove.getHeaders().getId()); @@ -416,27 +412,6 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa return getMessageGroup(groupId); } - /** - * {@inheritDoc} - */ - public MessageGroup markMessageFromGroup(Object groupId, Message messageToMark) { - - final long updatedDate = System.currentTimeMillis(); - final String groupKey = getKey(groupId); - final String messageId = getKey(messageToMark.getHeaders().getId()); - - jdbcTemplate.update(getQuery(MARK_MESSAGE_IN_GROUP), new PreparedStatementSetter() { - public void setValues(PreparedStatement ps) throws SQLException { - logger.debug("Marking message " + messageId + " in group with group key=" + groupKey); - ps.setTimestamp(1, new Timestamp(updatedDate)); - ps.setString(2, messageId); - ps.setString(3, groupKey); - ps.setString(4, region); - } - }); - return getMessageGroup(groupId); - } - public void removeMessageGroup(Object groupId) { final String groupKey = getKey(groupId); @@ -550,6 +525,24 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa private String getKey(Object input) { return input == null ? null : UUIDConverter.getUUID(input).toString(); } + + private long getGroupCreatedDate(String groupKey) { + final AtomicReference date = new AtomicReference(); + this.jdbcTemplate.query(getQuery(GET_GROUP_CREATED_DATE), new Object[] { groupKey, region }, + + new RowCallbackHandler() { + public void processRow(ResultSet rs) throws SQLException { + date.set(rs.getTimestamp("CREATED_DATE").getTime()); + } + }); + Long returnedDate = date.get(); + if (returnedDate == null){ + return 0; + } + else { + return returnedDate; + } + } /** * Convenience class to be used to unpack a message from a result set row. Uses column named in the result set to diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java index 09dee78aa1..d5d2aad8ab 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java @@ -241,17 +241,6 @@ public class JdbcMessageStoreTests { assertEquals(1, messageStore.getMessageCountForAllMessageGroups()); } - @Test - @Transactional - public void testMarkedMessageGroupSizes() throws Exception { - String groupId = "X"; - Message message = MessageBuilder.withPayload("foo").build(); - messageStore.addMessageToGroup(groupId, message); - assertEquals(0, messageStore.getMarkedMessageCountForAllMessageGroups()); - messageStore.markMessageGroup(messageStore.getMessageGroup(groupId)); - assertEquals(1, messageStore.getMarkedMessageCountForAllMessageGroups()); - } - @Test @Transactional public void testOrderInMessageGroup() throws Exception { @@ -281,5 +270,54 @@ public class JdbcMessageStoreTests { MessageGroup group = messageStore.getMessageGroup(groupId); assertEquals(0, group.size()); } + + @Test + @Transactional + public void testExpireMessageGroupOnCreateOnly() throws Exception { + String groupId = "X"; + Message message = MessageBuilder.withPayload("foo").setCorrelationId(groupId).build(); + messageStore.addMessageToGroup(groupId, message); + messageStore.registerMessageGroupExpiryCallback(new MessageGroupCallback() { + public void execute(MessageGroupStore messageGroupStore, MessageGroup group) { + messageGroupStore.removeMessageGroup(group.getGroupId()); + } + }); + Thread.sleep(1000); + messageStore.expireMessageGroups(2000); + MessageGroup group = messageStore.getMessageGroup(groupId); + assertEquals(1, group.size()); + Thread.sleep(2000); + messageStore.addMessageToGroup(groupId, MessageBuilder.withPayload("bar").setCorrelationId(groupId).build()); + messageStore.expireMessageGroups(2000); + group = messageStore.getMessageGroup(groupId); + assertEquals(0, group.size()); + } + + @Test + @Transactional + public void testExpireMessageGroupOnIdleOnly() throws Exception { + String groupId = "X"; + Message message = MessageBuilder.withPayload("foo").setCorrelationId(groupId).build(); + messageStore.setTimeoutOnIdle(true); + messageStore.addMessageToGroup(groupId, message); + messageStore.registerMessageGroupExpiryCallback(new MessageGroupCallback() { + public void execute(MessageGroupStore messageGroupStore, MessageGroup group) { + messageGroupStore.removeMessageGroup(group.getGroupId()); + } + }); + Thread.sleep(1000); + messageStore.expireMessageGroups(2000); + MessageGroup group = messageStore.getMessageGroup(groupId); + assertEquals(1, group.size()); + Thread.sleep(2000); + messageStore.addMessageToGroup(groupId, MessageBuilder.withPayload("bar").setCorrelationId(groupId).build()); + messageStore.expireMessageGroups(2000); + group = messageStore.getMessageGroup(groupId); + assertEquals(2, group.size()); + Thread.sleep(2000); + messageStore.expireMessageGroups(1000); + group = messageStore.getMessageGroup(groupId); + assertEquals(0, group.size()); + } } 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 2615b77caa..68d8cd7f84 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 @@ -74,6 +74,8 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore implements Me private final static String LAST_RELEASED_SEQUENCE_NUMBER = "_last_released_sequence"; private final static String GROUP_TIMESTAMP_KEY = "_group_timestamp"; + + private final static String GROUP_UPDATE_TIMESTAMP_KEY = "_group_update_timestamp"; private final static String PAYLOAD_TYPE_KEY = "_payloadType"; @@ -139,11 +141,13 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore implements Me List messageWrappers = this.template.find(whereGroupIdIs(groupId), MessageWrapper.class, this.collectionName); List> messages = new ArrayList>(); long timestamp = 0; + long lastmodified = 0; int lastReleasedSequenceNumber = 0; boolean completeGroup = false; if (messageWrappers.size() > 0){ MessageWrapper messageWrapper = messageWrappers.get(0); timestamp = messageWrapper.getGroupTimestamp(); + lastmodified = messageWrapper.getLastModified(); completeGroup = messageWrapper.isCompletedGroup(); lastReleasedSequenceNumber = messageWrapper.getLastReleasedSequenceNumber(); } @@ -151,7 +155,9 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore implements Me for (MessageWrapper messageWrapper : messageWrappers) { messages.add(messageWrapper.getMessage()); } + SimpleMessageGroup messageGroup = new SimpleMessageGroup(messages, groupId, timestamp, completeGroup); + messageGroup.setLastModified(lastmodified); if (lastReleasedSequenceNumber > 0){ messageGroup.setLastReleasedMessageSequenceNumber(lastReleasedSequenceNumber); } @@ -163,10 +169,22 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore implements Me Assert.notNull(groupId, "'groupId' must not be null"); Assert.notNull(message, "'message' must not be null"); MessageGroup messageGroup = this.getMessageGroup(groupId); + + long messageGroupTimestamp = messageGroup.getTimestamp(); + long lastModified = messageGroup.getLastModified(); + + if (messageGroupTimestamp == 0){ + messageGroupTimestamp = System.currentTimeMillis(); + lastModified = messageGroupTimestamp; + } + else { + lastModified = System.currentTimeMillis(); + } MessageWrapper wrapper = new MessageWrapper(message); wrapper.setGroupId(groupId); - wrapper.setGroupTimestamp(messageGroup.getTimestamp()); + wrapper.setGroupTimestamp(messageGroupTimestamp); + wrapper.setLastModified(lastModified); wrapper.setCompletedGroup(messageGroup.isComplete()); wrapper.setLastReleasedSequenceNumber(messageGroup.getLastReleasedMessageSequenceNumber()); @@ -234,7 +252,9 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore implements Me } private static Query whereGroupIdIs(Object groupId) { - return new Query(where(GROUP_ID_KEY).is(groupId)); + Query q = new Query(where(GROUP_ID_KEY).is(groupId)); + q.sort().on(GROUP_UPDATE_TIMESTAMP_KEY, Order.DESCENDING); + return q; } private static Query whereGroupIdExists() { @@ -274,6 +294,7 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore implements Me boolean groupComplete = false; long groupTimestamp = 0; + long lastModified = 0; int lastReleasedSequenceNumber = 0; if (source instanceof MessageWrapper) { MessageWrapper wrapper = (MessageWrapper) source; @@ -282,6 +303,7 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore implements Me groupComplete = wrapper.isCompletedGroup(); lastReleasedSequenceNumber = wrapper.getLastReleasedSequenceNumber(); groupTimestamp = wrapper.getGroupTimestamp(); + lastModified = wrapper.getLastModified(); } else { Class sourceType = (source != null) ? source.getClass() : null; @@ -294,6 +316,7 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore implements Me target.put(GROUP_COMPLETE_KEY, groupComplete); target.put(LAST_RELEASED_SEQUENCE_NUMBER, lastReleasedSequenceNumber); target.put(GROUP_TIMESTAMP_KEY, groupTimestamp); + target.put(GROUP_UPDATE_TIMESTAMP_KEY, lastModified); } super.write(message, target); @@ -324,6 +347,7 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore implements Me innerMap.put(MessageHeaders.ID, UUID.fromString((String) headers.get(MessageHeaders.ID))); innerMap.put(MessageHeaders.TIMESTAMP, headers.get(MessageHeaders.TIMESTAMP)); Long groupTimestamp = (Long)source.get(GROUP_TIMESTAMP_KEY); + Long lastModified = (Long)source.get(GROUP_UPDATE_TIMESTAMP_KEY); Integer lastReleasedSequenceNumber = (Integer)source.get(LAST_RELEASED_SEQUENCE_NUMBER); Boolean completeGroup = (Boolean)source.get(GROUP_COMPLETE_KEY); @@ -335,6 +359,9 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore implements Me if (groupTimestamp != null){ wrapper.setGroupTimestamp(groupTimestamp); } + if (lastModified != null){ + wrapper.setLastModified(lastModified); + } if (lastReleasedSequenceNumber != null){ wrapper.setLastReleasedSequenceNumber(lastReleasedSequenceNumber); } @@ -375,6 +402,8 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore implements Me private volatile long groupTimestamp; + private volatile long lastModified; + private volatile int lastReleasedSequenceNumber; private volatile boolean completedGroup; @@ -410,6 +439,14 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore implements Me public void setGroupTimestamp(long groupTimestamp) { this.groupTimestamp = groupTimestamp; } + + public long getLastModified() { + return lastModified; + } + + public void setLastModified(long lastModified) { + this.lastModified = lastModified; + } public void setLastReleasedSequenceNumber(int lastReleasedSequenceNumber) { this.lastReleasedSequenceNumber = lastReleasedSequenceNumber; diff --git a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/MongoDbMessageGroupStoreTests.java b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/MongoDbMessageGroupStoreTests.java index 0f17cbdef1..7306da7906 100644 --- a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/MongoDbMessageGroupStoreTests.java +++ b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/MongoDbMessageGroupStoreTests.java @@ -19,6 +19,7 @@ import java.util.Iterator; 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; @@ -74,6 +75,34 @@ public class MongoDbMessageGroupStoreTests extends MongoDbAvailableTests { assertNull(retrievedMessage.getHeaders().get("message_group")); } + @Test + @MongoDbAvailable + public void testMessageGroupUpdatedDateChangesWithEachAddedMessage() throws Exception{ + MongoDbFactory mongoDbFactory = this.prepareMongoFactory(); + MongoDbMessageStore store = new MongoDbMessageStore(mongoDbFactory); + + MessageGroup messageGroup = store.getMessageGroup(1); + Message message = new GenericMessage("Hello"); + messageGroup = store.addMessageToGroup(1, message); + assertEquals(1, messageGroup.size()); + long createdTimestamp = messageGroup.getTimestamp(); + long updatedTimestamp = messageGroup.getLastModified(); + assertEquals(createdTimestamp, updatedTimestamp); + Thread.sleep(1000); + message = new GenericMessage("Hello again"); + messageGroup = store.addMessageToGroup(1, message); + createdTimestamp = messageGroup.getTimestamp(); + updatedTimestamp = messageGroup.getLastModified(); + assertTrue(updatedTimestamp > createdTimestamp); + assertEquals(2, messageGroup.size()); + + // make sure the store is properly rebuild from MongoDB + store = new MongoDbMessageStore(mongoDbFactory); + + messageGroup = store.getMessageGroup(1); + assertEquals(2, messageGroup.size()); + } + @Test @MongoDbAvailable public void testMessageGroupMarkingMessage() throws Exception{ diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisMessageGroupStoreTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisMessageGroupStoreTests.java index 4c2213e761..94f5815190 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisMessageGroupStoreTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisMessageGroupStoreTests.java @@ -62,6 +62,33 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests { assertEquals(0, messageGroup.size()); } + @Test + @RedisAvailable + public void testMessageGroupUpdatedDateChangesWithEachAddedMessage() throws Exception{ + JedisConnectionFactory jcf = this.getConnectionFactoryForTest(); + RedisMessageStore store = new RedisMessageStore(jcf); + + MessageGroup messageGroup = store.getMessageGroup(1); + Message message = new GenericMessage("Hello"); + messageGroup = store.addMessageToGroup(1, message); + assertEquals(1, messageGroup.size()); + long createdTimestamp = messageGroup.getTimestamp(); + long updatedTimestamp = messageGroup.getLastModified(); + assertEquals(createdTimestamp, updatedTimestamp); + Thread.sleep(1000); + message = new GenericMessage("Hello"); + messageGroup = store.addMessageToGroup(1, message); + createdTimestamp = messageGroup.getTimestamp(); + updatedTimestamp = messageGroup.getLastModified(); + assertTrue(updatedTimestamp > createdTimestamp); + + // make sure the store is properly rebuild from Redis + store = new RedisMessageStore(jcf); + + messageGroup = store.getMessageGroup(1); + assertEquals(2, messageGroup.size()); + } + @Test @RedisAvailable public void testMessageGroupWithAddedMessage() throws Exception{