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

INT-3037 Fix JDBC MS Discard After Completion

Browse Source 
INT-3037 - commented out sizing component

unit test to support

cleanup

updated commit as per Artem's comments May 29
This commit is contained in:
willschipp
2013-05-28 14:58:37 -04:00
committed by Gary Russell
parent fcdef894bd
commit f6ef5d0133
2 changed files with 47 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2013 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
@@ -31,6 +31,7 @@ import javax.sql.DataSource;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.core.serializer.Deserializer;
import org.springframework.core.serializer.Serializer;
@@ -75,6 +76,7 @@ import org.springframework.util.StringUtils;
* @author Oleg Zhurakousky
* @author Matt Stine
* @author Gunnar Hillert
* @author Will Schipp
*
* @since 2.0
*/
@@ -419,10 +421,6 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
List<Message<?>> messages = jdbcTemplate.query(getQuery(Query.LIST_MESSAGES_BY_GROUP_KEY), new Object[] { key, region }, mapper);
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 {

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2013 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.
@@ -65,6 +65,7 @@ import org.springframework.transaction.annotation.Transactional;
* @author Oleg Zhurakousky
* @author Gunnar Hillert
* @author Artem Bilan
* @author Will Schipp
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@@ -424,9 +425,49 @@ public class JdbcMessageStoreTests {
LOG.info("messageFromGroup1: " + messageFromGroup1.getHeaders().getId() + "; Sequence #: " + messageFromGroup1.getHeaders().getSequenceNumber());
LOG.info("messageFromGroup2: " + messageFromGroup2.getHeaders().getId() + "; Sequence #: " + messageFromGroup2.getHeaders().getSequenceNumber());
assertEquals(Integer.valueOf(1), (Integer) messageFromGroup1.getHeaders().get(MessageHeaders.SEQUENCE_NUMBER));
assertEquals(Integer.valueOf(2), (Integer) messageFromGroup2.getHeaders().get(MessageHeaders.SEQUENCE_NUMBER));
assertEquals(Integer.valueOf(1), messageFromGroup1.getHeaders().get(MessageHeaders.SEQUENCE_NUMBER));
assertEquals(Integer.valueOf(2), messageFromGroup2.getHeaders().get(MessageHeaders.SEQUENCE_NUMBER));
}
@Test
@Transactional
public void testCompletedNotExpiredGroupINT3037() throws Exception {
/*
* based on the aggregator scenario as follows;
*
* send three messages in
* 1 of 2
* 2 of 2
* 2 of 2 (last again)
*
* expected behavior is that the LAST message (2 of 2 repeat) should be on the discard channel
* (discard behavior performed by the AbstractCorrelatingMessageHandler.handleMessageInternal)
*/
final JdbcMessageStore messageStore = new JdbcMessageStore(dataSource);
//init
String groupId = "group";
//build the messages
Message<?> oneOfTwo = MessageBuilder.withPayload("hello").setSequenceNumber(1).setSequenceSize(2).setCorrelationId(groupId).build();
Message<?> twoOfTwo = MessageBuilder.withPayload("world").setSequenceNumber(2).setSequenceSize(2).setCorrelationId(groupId).build();
//add to the messageStore
messageStore.addMessageToGroup(groupId, oneOfTwo);
messageStore.addMessageToGroup(groupId, twoOfTwo);
//check that 2 messages are there
assertTrue(messageStore.getMessageGroupCount() == 1);
assertTrue(messageStore.getMessageCount() == 2);
//retrieve the group (like in the aggregator)
MessageGroup messageGroup = messageStore.getMessageGroup(groupId);
//'complete' the group
messageStore.completeGroup(messageGroup.getGroupId());
//now clear the messages
for (Message<?> message : messageGroup.getMessages()) {
messageStore.removeMessageFromGroup(groupId, message);
}//end for
//'add' the other message --> emulated by getting the messageGroup
messageGroup = messageStore.getMessageGroup(groupId);
//should be marked 'complete' --> old behavior it would not
assertTrue(messageGroup.isComplete());
}
}