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

Merge pull request #631 from olegz/INT-2757

Browse Source
This commit is contained in:
Gary Russell
2012-09-20 14:34:45 +01:00
parent 06e2f61969 b4af191f4f
commit 26e2a45ee6
2 changed files with 52 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -89,6 +89,8 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
REMOVE_MESSAGE_FROM_GROUP("DELETE from %PREFIX%GROUP_TO_MESSAGE where GROUP_KEY=? and MESSAGE_ID=?"),
REMOVE_GROUP_TO_MESSAGE_JOIN("DELETE from %PREFIX%GROUP_TO_MESSAGE where GROUP_KEY=?"),
COUNT_ALL_MESSAGES_IN_GROUPS("SELECT COUNT(MESSAGE_ID) from %PREFIX%GROUP_TO_MESSAGE"),
COUNT_ALL_MESSAGES_IN_GROUP("SELECT COUNT(MESSAGE_ID) from %PREFIX%GROUP_TO_MESSAGE where GROUP_KEY=?"),
@@ -403,7 +405,11 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
List<Message<?>> messages = jdbcTemplate.query(getQuery(Query.LIST_MESSAGES_BY_GROUP_KEY), new Object[] { key, region }, mapper);
jdbcTemplate.query(getQuery(Query.GET_GROUP_INFO), new Object[] { key},
if (messages.size() == 0){
return new SimpleMessageGroup(groupId);
}
jdbcTemplate.query(getQuery(Query.GET_GROUP_INFO), new Object[] { key},
new RowCallbackHandler() {
public void processRow(ResultSet rs) throws SQLException {
updateDate.set(rs.getTimestamp("UPDATED_DATE"));
@@ -416,11 +422,14 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
}
});
if (messages.size() == 0){
if (createDate.get() == null && updateDate.get() == null) {
if (logger.isWarnEnabled()) {
for (Message<?> message : messages) {
logger.warn("Missing group row for message id: " + message.getHeaders().getId());
}
}
return new SimpleMessageGroup(groupId);
}
Assert.state(createDate.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 = createDate.get().getTime();
boolean complete = completeFlag.get().booleanValue();
@@ -460,6 +469,15 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
this.removeMessage(messageIds);
}
jdbcTemplate.update(getQuery(Query.REMOVE_GROUP_TO_MESSAGE_JOIN), new PreparedStatementSetter() {
public void setValues(PreparedStatement ps) throws SQLException {
if (logger.isDebugEnabled()){
logger.debug("Removing relationships for the group with group key=" + groupKey);
}
ps.setString(1, groupKey);
}
});
jdbcTemplate.update(getQuery(Query.DELETE_MESSAGE_GROUP), new PreparedStatementSetter() {
public void setValues(PreparedStatement ps) throws SQLException {
if (logger.isDebugEnabled()){

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

@@ -16,6 +16,15 @@
package org.springframework.integration.jdbc;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNotSame;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.springframework.integration.test.matcher.PayloadAndHeaderMatcher.sameExceptIgnorableHeaders;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
@@ -41,20 +50,12 @@ import org.springframework.integration.store.MessageGroup;
import org.springframework.integration.store.MessageGroupCallback;
import org.springframework.integration.store.MessageGroupStore;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.util.UUIDConverter;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.transaction.annotation.Transactional;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNotSame;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.springframework.integration.test.matcher.PayloadAndHeaderMatcher.sameExceptIgnorableHeaders;
/**
* @author Dave Syer
* @author Mark Fisher
@@ -234,6 +235,24 @@ public class JdbcMessageStoreTests {
assertEquals(0, group.size());
}
@Test
@Transactional
public void testRemoveMessageGroup() throws Exception {
JdbcTemplate template = new JdbcTemplate(dataSource);
template.afterPropertiesSet();
String groupId = "X";
Message<String> message = MessageBuilder.withPayload("foo").setCorrelationId(groupId).build();
messageStore.addMessageToGroup(groupId, message);
messageStore.removeMessageGroup(groupId);
MessageGroup group = messageStore.getMessageGroup(groupId);
assertEquals(0, group.size());
String uuidGroupId = UUIDConverter.getUUID(groupId).toString();
assertTrue(template.queryForList(
"SELECT * from INT_GROUP_TO_MESSAGE where GROUP_KEY = '" + uuidGroupId + "'").size() == 0);
}
@Test
@Transactional
public void testCompleteMessageGroup() throws Exception {
@@ -279,7 +298,7 @@ public class JdbcMessageStoreTests {
@Transactional
public void testOrderInMessageGroup() throws Exception {
String groupId = "X";
messageStore.addMessageToGroup(groupId, MessageBuilder.withPayload("foo").setCorrelationId(groupId).build());
Thread.sleep(1);
messageStore.addMessageToGroup(groupId, MessageBuilder.withPayload("bar").setCorrelationId(groupId).build());