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

INT-2231

Browse Source 
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
This commit is contained in:
Oleg Zhurakousky
2011-11-21 04:54:13 -05:00
committed by Mark Fisher
parent d67736021a
commit 50d8caa1c3
10 changed files with 237 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files

97
spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java
View File

@@ -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<Message<?>> messages = new ArrayList<Message<?>>();
final AtomicReference<Date> date = new AtomicReference<Date>();
final AtomicReference<Date> updateDate = new AtomicReference<Date>();
final AtomicReference<Boolean> completeFlag = new AtomicReference<Boolean>();
final AtomicReference<Integer> lastReleasedSequenceRef = new AtomicReference<Integer>();
@@ -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<Long> date = new AtomicReference<Long>();
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

60
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java
View File

@@ -241,17 +241,6 @@ public class JdbcMessageStoreTests {
assertEquals(1, messageStore.getMessageCountForAllMessageGroups());
}
@Test
@Transactional
public void testMarkedMessageGroupSizes() throws Exception {
String groupId = "X";
Message<String> 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<String> 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<String> 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());
}
}