diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java index 66dab01514..b8c200b889 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java @@ -27,6 +27,7 @@ import org.springframework.integration.handler.AbstractMessageHandler; import org.springframework.integration.store.*; import org.springframework.util.Assert; +import java.util.Collection; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; @@ -48,16 +49,13 @@ import java.util.concurrent.ConcurrentMap; * @author Dave Syer * @since 2.0 */ +@SuppressWarnings({"SynchronizationOnLocalVariableOrMethodParameter"}) public class CorrelatingMessageHandler extends AbstractMessageHandler implements MessageProducer { private static final Log logger = LogFactory.getLog(CorrelatingMessageHandler.class); public static final long DEFAULT_SEND_TIMEOUT = 1000L; - public static final long DEFAULT_REAPER_INTERVAL = 1000L; - - public static final long DEFAULT_TIMEOUT = 60000L; - private MessageGroupStore messageStore; @@ -166,21 +164,14 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements if (group.canAdd(message)) { group = store(correlationKey, message); if (releaseStrategy.canRelease(group)) { + Collection completedMessages = null; try { - completeGroup(message, correlationKey, group); + completedMessages = completeGroup(message, correlationKey, group); } finally { // Always clean up even if there was an exception // processing messages - if (group.isComplete() || group.getSequenceSize() == 0) { - // The group is complete or else there is no - // sequence so there is no more state to track - remove(group); - } else { - // Mark these messages as processed, but do not - // remove the group from store - mark(group); - } + cleanUpForReleasedGroup(group, completedMessages); } } else if (group.isComplete()) { try { @@ -200,6 +191,22 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements } } + private void cleanUpForReleasedGroup(MessageGroup group, Collection completedMessages) { + if (group.isComplete() || group.getSequenceSize() == 0) { + // The group is complete or else there is no + // sequence so there is no more state to track + remove(group); + } else { + // Mark these messages as processed, but do not + // remove the group from store + if (completedMessages == null) { + mark(group); + } else { + mark(group, completedMessages); + } + } + } + private final boolean forceComplete(MessageGroup group) { Object correlationKey = group.getGroupId(); @@ -232,6 +239,13 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements messageStore.markMessageGroup(group); } + private void mark(MessageGroup group, Collection partialSequence) { + Object id = group.getGroupId(); + for (Message message : partialSequence) { + messageStore.markMessageFromGroup(id, message); + } + } + private void remove(MessageGroup group) { Object correlationKey = group.getGroupId(); messageStore.removeMessageGroup(correlationKey); @@ -271,13 +285,19 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements completeGroup(first, correlationKey, group); } - private void completeGroup(Message message, Object correlationKey, MessageGroup group) { + private Collection completeGroup(Message message, Object correlationKey, MessageGroup group) { if (logger.isDebugEnabled()) { logger.debug("Completing group with correlationKey [" + correlationKey + "]"); } Object result = outputProcessor.processMessageGroup(group); + Collection partialSequence = null; + if (result instanceof Collection) { + //Taking a risk here because of Type Erasure. This is covered in the processor contract + partialSequence = (Collection) result; + } this.sendReplies(result, message); + return partialSequence; } private void sendReplies(Object processorResult, Message message) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MessageGroupProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MessageGroupProcessor.java index 13481f2a81..6e69291dda 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MessageGroupProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MessageGroupProcessor.java @@ -17,7 +17,7 @@ import org.springframework.integration.store.MessageGroup; /** * A processor for correlated groups of messages. - * + * * @author Iwein Fuld * @see org.springframework.integration.aggregator.CorrelatingMessageHandler */ @@ -27,6 +27,8 @@ public interface MessageGroupProcessor { * Process the given MessageGroup. Implementations are free to return as few or as many messages based on the * invocation as needed. For example an aggregating processor will return only a single message representing the * group, while a resequencing processor will return all messages whose preceding sequence has been satisfied. + *

+ * If a multiple messages are returned the return value must be a Collection<Message>. */ Object processMessageGroup(MessageGroup group); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ResequencingMessageGroupProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ResequencingMessageGroupProcessor.java index fc89af08c8..e738e79aa5 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ResequencingMessageGroupProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ResequencingMessageGroupProcessor.java @@ -13,18 +13,14 @@ package org.springframework.integration.aggregator; -import java.util.ArrayList; -import java.util.Collection; -import java.util.Collections; -import java.util.Comparator; -import java.util.List; - import org.springframework.integration.Message; import org.springframework.integration.store.MessageGroup; +import java.util.*; + /** * This class implements all the strategy interfaces needed for a default resequencer. - * + * * @author Iwein Fuld * @author Dave Syer * @since 2.0 @@ -35,6 +31,7 @@ public class ResequencingMessageGroupProcessor implements MessageGroupProcessor /** * A comparator to use to order messages before processing. The default is to order by sequence number. + * * @param comparator the comparator to use to order messages */ public void setComparator(Comparator> comparator) { @@ -43,12 +40,28 @@ public class ResequencingMessageGroupProcessor implements MessageGroupProcessor public Object processMessageGroup(MessageGroup group) { Collection> messages = group.getUnmarked(); + if (messages.size() > 0) { List> sorted = new ArrayList>(messages); Collections.sort(sorted, this.comparator); - return sorted; + ArrayList partialSequence = new ArrayList(); + int previousSequence = extractSequenceNumber(sorted.get(0)); + int currentSequence = previousSequence; + for (Message message : sorted) { + previousSequence = currentSequence; + currentSequence = extractSequenceNumber(message); + if (currentSequence - 1 > previousSequence) { + //there is a gap in the sequence here + break; + } + partialSequence.add(message); + } + return partialSequence; } return null; } + private Integer extractSequenceNumber(Message message) { + return message.getHeaders().getSequenceNumber(); + } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/SequenceSizeReleaseStrategy.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/SequenceSizeReleaseStrategy.java index 1e4e29c350..b98eeaa71d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/SequenceSizeReleaseStrategy.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/SequenceSizeReleaseStrategy.java @@ -16,14 +16,14 @@ package org.springframework.integration.aggregator; +import org.springframework.integration.Message; +import org.springframework.integration.store.MessageGroup; + import java.util.ArrayList; import java.util.Collections; import java.util.Comparator; import java.util.List; -import org.springframework.integration.Message; -import org.springframework.integration.store.MessageGroup; - /** * An implementation of {@link ReleaseStrategy} that simply compares the * current size of the message list to the expected 'sequenceSize'. @@ -60,9 +60,8 @@ public class SequenceSizeReleaseStrategy implements ReleaseStrategy { if (releasePartialSequences) { List> sorted = new ArrayList>(messages.getUnmarked()); Collections.sort(sorted, comparator); - int head = sorted.get(sorted.size() - 1).getHeaders().getSequenceNumber(); int tail = sorted.get(0).getHeaders().getSequenceNumber() - 1; - return tail == messages.getMarked().size() && head - tail == sorted.size(); + return tail == messages.getMarked().size(); } return messages.isComplete(); } 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 5dd7bc2dd1..c5186e98e2 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 @@ -59,4 +59,8 @@ public interface MessageGroup { */ long getTimestamp(); + /** + * Mark the given message in this group. If the message is not part of this group then this call has no effect. + */ + void mark(Message messageToMark); } \ No newline at end of file diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupStore.java b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupStore.java index 8c59f9d117..2c395a378b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupStore.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupStore.java @@ -48,13 +48,21 @@ public interface MessageGroupStore { */ MessageGroup markMessageGroup(MessageGroup group); + /** + * Persist a deletion on a single message from the group. The group is modified to reflect that 'messageToRemove' is no + * longer present in the group. + * @param key the groupId for the group containing the message + * @param messageToRemove the message to be removed + */ + MessageGroup removeMessageFromGroup(Object key, Message messageToRemove); + /** * Persist a mark on a single message from the group. The group is modified to reflect that 'messageToMark' is no * longer unmarked but became marked instead. * @param key the groupId for the group containing the message * @param messageToMark the message to be marked */ - MessageGroup removeMessageFromGroup(Object key, Message messageToMark); + MessageGroup markMessageFromGroup(Object key, Message messageToMark); /** * Remove the message group with this id. 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 cd0f297d55..e966a41de1 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 @@ -13,13 +13,13 @@ package org.springframework.integration.store; +import org.springframework.integration.Message; + import java.util.Collection; import java.util.Collections; import java.util.concurrent.BlockingQueue; import java.util.concurrent.LinkedBlockingQueue; -import org.springframework.integration.Message; - /** * Represents a mutable group of correlated messages that is bound to a certain {@link MessageStore} and group id. The * group will grow during its lifetime, when messages are added to it. This MessageGroup is thread safe. @@ -154,6 +154,9 @@ public class SimpleMessageGroup implements MessageGroup { return getOne().getHeaders().getSequenceSize(); } + /** + * {@inheritDoc}} + */ public void mark(Message messageToMark) { synchronized (lock) { unmarked.remove(messageToMark); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java b/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java index a1a79cb209..dec3c394a8 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java @@ -13,17 +13,17 @@ package org.springframework.integration.store; +import org.springframework.integration.Message; +import org.springframework.integration.MessagingException; +import org.springframework.integration.util.UpperBound; +import org.springframework.util.Assert; + import java.util.HashSet; import java.util.Iterator; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; -import org.springframework.integration.Message; -import org.springframework.integration.MessagingException; -import org.springframework.integration.util.UpperBound; -import org.springframework.util.Assert; - /** * Map-based implementation of {@link MessageStore} and {@link MessageGroupStore}. Enforces a maximum capacity for the * store. @@ -125,9 +125,15 @@ public class SimpleMessageStore extends AbstractMessageGroupStore implements Mes groupIdToMessageGroup.remove(groupId); } - public MessageGroup removeMessageFromGroup(Object key, Message messageToMark) { + public MessageGroup removeMessageFromGroup(Object key, Message messageToRemove) { SimpleMessageGroup group = getMessageGroupInternal(key); - group.remove(messageToMark); + group.remove(messageToRemove); + return group; + } + + public MessageGroup markMessageFromGroup(Object key, Message messageToMark) { + SimpleMessageGroup group = getMessageGroupInternal(key); + group.mark(messageToMark); return group; } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java index 1fd1898eb1..dedd6735a9 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java @@ -16,15 +16,6 @@ package org.springframework.integration.aggregator; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertNotNull; -import static org.junit.Assert.assertNull; - -import java.util.ArrayList; -import java.util.Arrays; -import java.util.Collections; -import java.util.Comparator; - import org.junit.Before; import org.junit.Test; import org.springframework.integration.Message; @@ -34,10 +25,19 @@ import org.springframework.integration.store.MessageGroupStore; import org.springframework.integration.store.SimpleMessageStore; import org.springframework.integration.support.MessageBuilder; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.Comparator; + +import static org.junit.Assert.*; +import static org.hamcrest.Matchers.*; + /** * @author Marius Bogoevici * @author Alex Peters * @author Dave Syer + * @author Iwein Fuld */ public class ResequencerTests { @@ -65,11 +65,11 @@ public class ResequencerTests { Message reply2 = replyChannel.receive(0); Message reply3 = replyChannel.receive(0); assertNotNull(reply1); - assertEquals(new Integer(1), reply1.getHeaders().getSequenceNumber()); + assertThat( reply1.getHeaders().getSequenceNumber(), is(1)); assertNotNull(reply2); - assertEquals(new Integer(2), reply2.getHeaders().getSequenceNumber()); + assertThat(reply2.getHeaders().getSequenceNumber(), is(2)); assertNotNull(reply3); - assertEquals(new Integer(3), reply3.getHeaders().getSequenceNumber()); + assertThat( reply3.getHeaders().getSequenceNumber(), is(3)); } @Test @@ -141,7 +141,8 @@ public class ResequencerTests { assertNull(reply3); // when sending the last message, the whole sequence must have been sent this.resequencer.handleMessage(message4); - reply3 = replyChannel.receive(0); Message reply4 = replyChannel.receive(0); + reply3 = replyChannel.receive(0); + Message reply4 = replyChannel.receive(0); assertNotNull(reply3); assertEquals(new Integer(3), reply3.getHeaders().getSequenceNumber()); assertNotNull(reply4); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ResequencingMessageGroupProcessorTest.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ResequencingMessageGroupProcessorTest.java new file mode 100644 index 0000000000..acd58551ea --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ResequencingMessageGroupProcessorTest.java @@ -0,0 +1,51 @@ +package org.springframework.integration.aggregator; + +import org.junit.Test; +import org.springframework.integration.Message; +import org.springframework.integration.store.SimpleMessageGroup; +import org.springframework.integration.support.MessageBuilder; + +import java.util.ArrayList; +import java.util.List; + +import static org.hamcrest.Matchers.*; +import static org.junit.Assert.assertThat; + +/** + * @author Iwein Fuld + */ +public class ResequencingMessageGroupProcessorTest { + + private ResequencingMessageGroupProcessor processor = new ResequencingMessageGroupProcessor(); + + @Test + public void shouldProcessSequence() { + Message prototypeMessage = MessageBuilder.withPayload("foo").setCorrelationId("x").setSequenceNumber(1).setSequenceSize(3).build(); + List> unmarked= new ArrayList>(); + Message message1 = MessageBuilder.fromMessage(prototypeMessage).setSequenceNumber(1).build(); + Message message2 = MessageBuilder.fromMessage(prototypeMessage).setSequenceNumber(2).build(); + Message message3 = MessageBuilder.fromMessage(prototypeMessage).setSequenceNumber(3).build(); + unmarked.add(message1); + unmarked.add(message2); + unmarked.add(message3); + SimpleMessageGroup group = new SimpleMessageGroup(unmarked,"x"); + List processedMessages = (List) processor.processMessageGroup(group); + assertThat(processedMessages, hasItems(message1, message2, message3)); + } + + @Test + public void shouldPartiallProcessIncompleteSequence() { + Message prototypeMessage = MessageBuilder.withPayload("foo").setCorrelationId("x").setSequenceNumber(1).setSequenceSize(4).build(); + List> unmarked= new ArrayList>(); + Message message2 = MessageBuilder.fromMessage(prototypeMessage).setSequenceNumber(4).build(); + Message message1 = MessageBuilder.fromMessage(prototypeMessage).setSequenceNumber(1).build(); + Message message3 = MessageBuilder.fromMessage(prototypeMessage).setSequenceNumber(3).build(); + unmarked.add(message1); + unmarked.add(message2); + unmarked.add(message3); + SimpleMessageGroup group = new SimpleMessageGroup(unmarked,"x"); + List processedMessages = (List) processor.processMessageGroup(group); + assertThat(processedMessages, hasItems(message1)); + assertThat(processedMessages.size(), is(1)); + } +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/SequenceSizeReleaseStrategyTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/SequenceSizeReleaseStrategyTests.java index 939fb20a0e..40f0dbba48 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/SequenceSizeReleaseStrategyTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/SequenceSizeReleaseStrategyTests.java @@ -86,7 +86,9 @@ public class SequenceSizeReleaseStrategyTests { SequenceSizeReleaseStrategy releaseStrategy = new SequenceSizeReleaseStrategy(); releaseStrategy.setReleasePartialSequences(true); - assertTrue(releaseStrategy.canRelease(groupWithLastAndFirstMessagesOfIncompleteSequence())); + boolean canRelease = releaseStrategy.canRelease(groupWithLastAndFirstMessagesOfIncompleteSequence()); + + assertTrue(canRelease); } private MessageGroup groupWithLastAndFirstMessagesOfIncompleteSequence() { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/ResequencerParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/ResequencerParserTests.java index 713b0d51ba..82f2f1c196 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/ResequencerParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/ResequencerParserTests.java @@ -13,33 +13,23 @@ package org.springframework.integration.config; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertNotNull; -import static org.junit.Assert.assertNull; -import static org.junit.Assert.assertTrue; -import static org.springframework.integration.test.util.TestUtils.getPropertyValue; - -import java.util.ArrayList; -import java.util.Comparator; -import java.util.List; - import org.junit.Before; import org.junit.Test; import org.springframework.context.ApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.Message; import org.springframework.integration.MessageChannel; -import org.springframework.integration.aggregator.CorrelatingMessageHandler; -import org.springframework.integration.aggregator.CorrelationStrategy; -import org.springframework.integration.aggregator.MethodInvokingCorrelationStrategy; -import org.springframework.integration.aggregator.MethodInvokingReleaseStrategy; -import org.springframework.integration.aggregator.ResequencingMessageGroupProcessor; +import org.springframework.integration.aggregator.*; import org.springframework.integration.channel.NullChannel; -import org.springframework.integration.core.PollableChannel; import org.springframework.integration.endpoint.EventDrivenConsumer; import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.test.util.TestUtils; +import java.util.Comparator; + +import static org.junit.Assert.*; +import static org.springframework.integration.test.util.TestUtils.getPropertyValue; + /** * @author Marius Bogoevici * @author Mark Fisher @@ -54,28 +44,6 @@ public class ResequencerParserTests { this.context = new ClassPathXmlApplicationContext("resequencerParserTests.xml", this.getClass()); } - @Test - public void testResequencing() { - MessageChannel inputChannel = (MessageChannel) context.getBean("inputChannel"); - PollableChannel outputChannel = (PollableChannel) context.getBean("outputChannel"); - List> outboundMessages = new ArrayList>(); - outboundMessages.add(createMessage("123", "id1", 3, 3, outputChannel)); - outboundMessages.add(createMessage("789", "id1", 3, 1, outputChannel)); - outboundMessages.add(createMessage("456", "id1", 3, 2, outputChannel)); - for (Message message : outboundMessages) { - inputChannel.send(message); - } - Message message1 = outputChannel.receive(500); - Message message2 = outputChannel.receive(500); - Message message3 = outputChannel.receive(500); - assertNotNull(message1); - assertEquals(new Integer(1), message1.getHeaders().getSequenceNumber()); - assertNotNull(message2); - assertEquals(new Integer(2), message2.getHeaders().getSequenceNumber()); - assertNotNull(message3); - assertEquals(new Integer(3), message3.getHeaders().getSequenceNumber()); - } - @Test public void testDefaultResequencerProperties() { EventDrivenConsumer endpoint = (EventDrivenConsumer) context.getBean("defaultResequencer"); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/store/MessageStoreTests.java b/spring-integration-core/src/test/java/org/springframework/integration/store/MessageStoreTests.java index 4c2eb24d96..cd41965383 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/store/MessageStoreTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/store/MessageStoreTests.java @@ -16,20 +16,15 @@ package org.springframework.integration.store; -import static org.junit.Assert.assertEquals; - -import java.util.ArrayList; -import java.util.Arrays; -import java.util.Collection; -import java.util.Iterator; -import java.util.List; - import org.junit.Test; - import org.springframework.integration.Message; import org.springframework.integration.message.GenericMessage; import org.springframework.test.util.ReflectionTestUtils; +import java.util.*; + +import static org.junit.Assert.assertEquals; + /** * @author Dave Syer */ @@ -86,7 +81,11 @@ public class MessageStoreTests { throw new UnsupportedOperationException(); } - public MessageGroup removeMessageFromGroup(Object key, Message messageToMark) { + public MessageGroup removeMessageFromGroup(Object key, Message messageToRemove) { + throw new UnsupportedOperationException(); + } + + public MessageGroup markMessageFromGroup(Object key, Message messageToMark) { throw new UnsupportedOperationException(); } 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 9b38c20ad2..7efa17c2d5 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 @@ -13,17 +13,6 @@ package org.springframework.integration.jdbc; -import java.sql.PreparedStatement; -import java.sql.ResultSet; -import java.sql.SQLException; -import java.sql.Timestamp; -import java.sql.Types; -import java.util.Iterator; -import java.util.List; -import java.util.UUID; - -import javax.sql.DataSource; - import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.commons.serializer.DeserializingConverter; @@ -38,16 +27,18 @@ import org.springframework.integration.store.MessageStore; import org.springframework.integration.store.SimpleMessageGroup; import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.util.UUIDConverter; -import org.springframework.jdbc.core.JdbcOperations; -import org.springframework.jdbc.core.JdbcTemplate; -import org.springframework.jdbc.core.PreparedStatementSetter; -import org.springframework.jdbc.core.RowMapper; -import org.springframework.jdbc.core.SingleColumnRowMapper; +import org.springframework.jdbc.core.*; import org.springframework.jdbc.support.lob.DefaultLobHandler; import org.springframework.jdbc.support.lob.LobHandler; import org.springframework.util.Assert; import org.springframework.util.StringUtils; +import javax.sql.DataSource; +import java.sql.*; +import java.util.Iterator; +import java.util.List; +import java.util.UUID; + /** * Implementation of {@link MessageStore} using a relational database via JDBC. SQL scripts to create the necessary * tables are packaged as org/springframework/integration/jdbc/schema-*.sql, where * is the @@ -330,11 +321,29 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa } - public MessageGroup removeMessageFromGroup(Object groupId, Message messageToMark) { + public MessageGroup removeMessageFromGroup(Object groupId, Message messageToRemove) { + final String groupKey = getKey(groupId); + final String messageId = getKey(messageToRemove.getHeaders().getId()); + + jdbcTemplate.update(getQuery(REMOVE_MESSAGE_FROM_GROUP), new PreparedStatementSetter() { + public void setValues(PreparedStatement ps) throws SQLException { + logger.debug("Removing message from group with group key=" + groupKey); + ps.setString(1, groupKey); + ps.setString(2, region); + ps.setString(3, messageId); + } + }); + return getMessageGroup(groupId); + } + + /** + * {@inheritDoc} + */ + public MessageGroup markMessageFromGroup(Object groupId, Message messageToMark) { final String groupKey = getKey(groupId); final String messageId = getKey(messageToMark.getHeaders().getId()); - jdbcTemplate.update(getQuery(REMOVE_MESSAGE_FROM_GROUP), new PreparedStatementSetter() { + jdbcTemplate.update(getQuery(MARK_MESSAGES_IN_GROUP), new PreparedStatementSetter() { public void setValues(PreparedStatement ps) throws SQLException { logger.debug("Removing message from group with group key=" + groupKey); ps.setString(1, groupKey);