diff --git a/org.springframework.integration.jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java b/org.springframework.integration.jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java index 3dbd12216e..32c47f09b5 100644 --- a/org.springframework.integration.jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java +++ b/org.springframework.integration.jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java @@ -21,7 +21,6 @@ import java.sql.ResultSet; import java.sql.SQLException; import java.sql.Timestamp; import java.sql.Types; -import java.util.Collection; import java.util.List; import java.util.UUID; @@ -29,7 +28,6 @@ import javax.sql.DataSource; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; - import org.springframework.integration.core.Message; import org.springframework.integration.jdbc.util.SerializationUtils; import org.springframework.integration.message.MessageBuilder; @@ -174,8 +172,8 @@ public class JdbcMessageStore implements MessageStore { Assert.state(jdbcTemplate != null, "A DataSource or JdbcTemplate must be provided"); } - public Message delete(UUID id) { - Message message = get(id); + public Message removeMessage(UUID id) { + Message message = getMessage(id); if (message == null) { return null; } @@ -187,7 +185,7 @@ public class JdbcMessageStore implements MessageStore { return null; } - public Message get(UUID id) { + public Message getMessage(UUID id) { List> list = jdbcTemplate.query(getQuery(GET_MESSAGE), new Object[] { getKey(id) }, mapper); if (list.isEmpty()) { return null; @@ -200,10 +198,10 @@ public class JdbcMessageStore implements MessageStore { new Object[] { getKey(correlationId) }, mapper); } - public Message put(final Message message) { + public Message addMessage(final Message message) { if (message.getHeaders().containsKey(SAVED_KEY)) { @SuppressWarnings("unchecked") - Message saved = (Message) get(message.getHeaders().getId()); + Message saved = (Message) getMessage(message.getHeaders().getId()); if (saved != null) { if (saved.equals(message)) { return message; @@ -230,23 +228,6 @@ public class JdbcMessageStore implements MessageStore { return result; } - public void put(Object correlationId, Message message) { - // TODO Auto-generated method stub - } - - public void put(Object correlationId, Collection> messages) { - // TODO Auto-generated method stub - } - - public Message mark(Object correlationId, UUID messageId) { - // TODO Auto-generated method stub - return null; - } - - public void deleteAll(Object correlationId) { - // TODO Auto-generated method stub - } - private String getKey(Object input) { return input == null ? null : UUIDConverter.getUUID(input).toString(); } diff --git a/org.springframework.integration.jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java b/org.springframework.integration.jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java index d6c9ca3f53..b5c5cdc358 100644 --- a/org.springframework.integration.jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java +++ b/org.springframework.integration.jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java @@ -38,7 +38,7 @@ public class JdbcMessageStoreTests { @Test @Transactional public void testGetNonExistent() throws Exception { - Message result = messageStore.get(UUID.randomUUID()); + Message result = messageStore.getMessage(UUID.randomUUID()); assertNull(result); } @@ -46,9 +46,9 @@ public class JdbcMessageStoreTests { @Transactional public void testAddAndGet() throws Exception { Message message = MessageBuilder.withPayload("foo").build(); - Message saved = messageStore.put(message); - assertNull(messageStore.get(message.getHeaders().getId())); - Message result = messageStore.get(saved.getHeaders().getId()); + Message saved = messageStore.addMessage(message); + assertNull(messageStore.getMessage(message.getHeaders().getId())); + Message result = messageStore.getMessage(saved.getHeaders().getId()); assertNotNull(result); assertThat(saved, sameExceptIgnorableHeaders(result)); assertNotNull(result.getHeaders().get(JdbcMessageStore.SAVED_KEY)); @@ -60,18 +60,18 @@ public class JdbcMessageStoreTests { public void testAddAndUpdate() throws Exception { Message message = MessageBuilder.withPayload("foo").setCorrelationId( "X").build(); - message = messageStore.put(message); + message = messageStore.addMessage(message); message = MessageBuilder.fromMessage(message).setCorrelationId("Y").build(); - message = messageStore.put(message); - assertEquals("Y", messageStore.get(message.getHeaders().getId()).getHeaders().getCorrelationId()); + message = messageStore.addMessage(message); + assertEquals("Y", messageStore.getMessage(message.getHeaders().getId()).getHeaders().getCorrelationId()); } @Test @Transactional public void testAddAndUpdateAlreadySaved() throws Exception { Message message = MessageBuilder.withPayload("foo").build(); - message = messageStore.put(message); - Message result = messageStore.put(message); + message = messageStore.addMessage(message); + Message result = messageStore.addMessage(message); assertEquals(message, result); } @@ -79,12 +79,12 @@ public class JdbcMessageStoreTests { @Transactional public void testAddAndUpdateAlreadySavedAndCopied() throws Exception { Message message = MessageBuilder.withPayload("foo").build(); - Message saved = messageStore.put(message); + Message saved = messageStore.addMessage(message); Message copy = MessageBuilder.fromMessage(saved).build(); - Message result = messageStore.put(copy); + Message result = messageStore.addMessage(copy); assertNotSame(copy, result); assertThat(saved, sameExceptIgnorableHeaders(result, JdbcMessageStore.CREATED_DATE_KEY)); - assertNotNull(messageStore.get(saved.getHeaders().getId())); + assertNotNull(messageStore.getMessage(saved.getHeaders().getId())); } @Test @@ -92,7 +92,7 @@ public class JdbcMessageStoreTests { public void testAddAndListByCorrelationId() throws Exception { String correlationId = "X"; Message message = MessageBuilder.withPayload("foo").setCorrelationId(correlationId).build(); - messageStore.put(message); + messageStore.addMessage(message); assertEquals(1, messageStore.list(correlationId).size()); } @@ -100,8 +100,8 @@ public class JdbcMessageStoreTests { @Transactional public void testAddAndDelete() throws Exception { Message message = MessageBuilder.withPayload("foo").build(); - message = messageStore.put(message); - assertNotNull(messageStore.delete(message.getHeaders().getId())); + message = messageStore.addMessage(message); + assertNotNull(messageStore.removeMessage(message.getHeaders().getId())); } } diff --git a/org.springframework.integration.jmx/.classpath b/org.springframework.integration.jmx/.classpath index cb112c7d0a..d8b3f86eb6 100644 --- a/org.springframework.integration.jmx/.classpath +++ b/org.springframework.integration.jmx/.classpath @@ -3,7 +3,7 @@ - + diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java index a8b8dae4f3..aa78446806 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java @@ -23,6 +23,7 @@ import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.core.MessageHeaders; import org.springframework.integration.message.MessageBuilder; +import org.springframework.integration.store.MessageGroup; import org.springframework.util.Assert; import java.util.*; diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java index e0fce3ea33..24c56bbb2d 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java @@ -13,7 +13,6 @@ package org.springframework.integration.aggregator; -import java.util.Collection; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; @@ -28,6 +27,8 @@ import org.springframework.integration.core.MessageHeaders; import org.springframework.integration.core.MessageProducer; import org.springframework.integration.handler.AbstractMessageHandler; import org.springframework.integration.message.MessageBuilder; +import org.springframework.integration.store.MessageGroup; +import org.springframework.integration.store.MessageGroupStore; import org.springframework.integration.store.MessageStore; import org.springframework.integration.store.SimpleMessageStore; import org.springframework.scheduling.TaskScheduler; @@ -58,7 +59,7 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements public static final long DEFAULT_TIMEOUT = 60000L; - private final MessageStore store; + private final MessageGroupStore store; private final MessageGroupProcessor outputProcessor; @@ -76,7 +77,7 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements private final ConcurrentMap locks = new ConcurrentHashMap(); - public CorrelatingMessageHandler(MessageGroupProcessor processor, MessageStore store, + public CorrelatingMessageHandler(MessageGroupProcessor processor, MessageGroupStore store, CorrelationStrategy correlationStrategy, ReleaseStrategy releaseStrategy) { Assert.notNull(store); Assert.notNull(processor); @@ -88,7 +89,7 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements this.channelTemplate.setSendTimeout(DEFAULT_SEND_TIMEOUT); } - public CorrelatingMessageHandler(MessageGroupProcessor processor, MessageStore store) { + public CorrelatingMessageHandler(MessageGroupProcessor processor, MessageGroupStore store) { this(processor, store, null, null); } @@ -160,8 +161,7 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements Object lock = getLock(correlationKey); synchronized (lock) { - Collection> messages = store.list(correlationKey); - MessageGroup group = new MessageGroup(messages, correlationKey); + MessageGroup group = store.getMessageGroup(correlationKey); if (group.add(message)) { @@ -209,12 +209,11 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements Object lock = getLock(correlationKey); synchronized (lock) { - Collection> all = store.list(correlationKey); - MessageGroup group = new MessageGroup(all, correlationKey); - if (all.size() > 0) { + MessageGroup group = store.getMessageGroup(correlationKey); + if (group.size() > 0) { // last chance for normal completion if (releaseStrategy.canRelease(group)) { - outputProcessor.processAndSend(group, channelTemplate, resolveReplyChannel(all.iterator().next(), + outputProcessor.processAndSend(group, channelTemplate, resolveReplyChannel(group.getOne(), this.outputChannel)); remove(group); } @@ -224,15 +223,15 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements logger.info("Processing partially complete messages for key [" + correlationKey + "] to: " + outputChannel); } - outputProcessor.processAndSend(group, channelTemplate, resolveReplyChannel(all.iterator() - .next(), this.outputChannel)); + outputProcessor.processAndSend(group, channelTemplate, resolveReplyChannel(group.getOne(), + this.outputChannel)); } else { if (logger.isInfoEnabled()) { logger.info("Discarding partially complete messages for key [" + correlationKey + "] to: " + discardChannel); } - for (Message message : all) { + for (Message message : group.getUnmarked()) { discardChannel.send(message); } } @@ -250,19 +249,17 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements } private void mark(MessageGroup group) { - for (Message message : group.getUnmarked()) { - store.mark(group.getCorrelationKey(), message.getHeaders().getId()); - } + store.mark(group); } private void remove(MessageGroup group) { Object correlationKey = group.getCorrelationKey(); - store.deleteAll(correlationKey); + store.deleteMessageGroup(correlationKey); locks.remove(correlationKey); } private void store(Object correlationKey, Message message) { - store.put(correlationKey, message); + store.addMessageToGroup(correlationKey, message); } } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/DefaultAggregatingMessageGroupProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/DefaultAggregatingMessageGroupProcessor.java index 2efdbdf980..fd5dbf49b8 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/DefaultAggregatingMessageGroupProcessor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/DefaultAggregatingMessageGroupProcessor.java @@ -21,6 +21,7 @@ import java.util.Collection; import java.util.List; import org.springframework.integration.core.Message; +import org.springframework.integration.store.MessageGroup; import org.springframework.util.Assert; /** diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageCountReleaseStrategy.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageCountReleaseStrategy.java index 63dadfd5c9..7f75aa063c 100755 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageCountReleaseStrategy.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageCountReleaseStrategy.java @@ -13,6 +13,8 @@ package org.springframework.integration.aggregator; +import org.springframework.integration.store.MessageGroup; + /** * A {@link ReleaseStrategy} that releases only the first n messages, where n is a threshold. * diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageGroupProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageGroupProcessor.java index 0279b3784a..0d777eb4e4 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageGroupProcessor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageGroupProcessor.java @@ -2,6 +2,7 @@ package org.springframework.integration.aggregator; import org.springframework.integration.channel.MessageChannelTemplate; import org.springframework.integration.core.MessageChannel; +import org.springframework.integration.store.MessageGroup; /** * A processor for correlated groups of messages. When a message group is complete it is passed to the diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessor.java index c69ded8b73..aa09896ad0 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessor.java @@ -29,6 +29,7 @@ import org.springframework.core.annotation.AnnotationUtils; import org.springframework.integration.annotation.Aggregator; import org.springframework.integration.annotation.Header; import org.springframework.integration.core.Message; +import org.springframework.integration.store.MessageGroup; import org.springframework.util.Assert; import org.springframework.util.ReflectionUtils; diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/PassThroughMessageGroupProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/PassThroughMessageGroupProcessor.java index dbba086239..fb3b892f85 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/PassThroughMessageGroupProcessor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/PassThroughMessageGroupProcessor.java @@ -3,6 +3,7 @@ package org.springframework.integration.aggregator; import org.springframework.integration.channel.MessageChannelTemplate; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; +import org.springframework.integration.store.MessageGroup; /** * This implementation of MessageGroupProcessor will forward all messages inside the group to the given output channel. diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/ReleaseStrategy.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/ReleaseStrategy.java index 98a8610227..5b3d5f6855 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/ReleaseStrategy.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/ReleaseStrategy.java @@ -16,6 +16,8 @@ package org.springframework.integration.aggregator; +import org.springframework.integration.store.MessageGroup; + /** diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/ReleaseStrategyAdapter.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/ReleaseStrategyAdapter.java index 9bfcc12245..0165417e44 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/ReleaseStrategyAdapter.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/ReleaseStrategyAdapter.java @@ -18,6 +18,7 @@ package org.springframework.integration.aggregator; import java.lang.reflect.Method; +import org.springframework.integration.store.MessageGroup; import org.springframework.util.Assert; /** diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/Resequencer.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/Resequencer.java index a483e4ec55..686dd5f4a7 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/Resequencer.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/Resequencer.java @@ -25,6 +25,7 @@ import java.util.List; import org.springframework.integration.channel.MessageChannelTemplate; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; +import org.springframework.integration.store.MessageGroup; /** * This class implements all the strategy interfaces needed for a default diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/SequenceSizeReleaseStrategy.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/SequenceSizeReleaseStrategy.java index 060ab98bda..28c18e4f2a 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/SequenceSizeReleaseStrategy.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/SequenceSizeReleaseStrategy.java @@ -16,6 +16,8 @@ package org.springframework.integration.aggregator; +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'. diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/store/MessageGroup.java b/org.springframework.integration/src/main/java/org/springframework/integration/store/MessageGroup.java new file mode 100644 index 0000000000..4ae3e36d36 --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/store/MessageGroup.java @@ -0,0 +1,43 @@ +package org.springframework.integration.store; + +import java.util.Collection; + +import org.springframework.integration.core.Message; + +public interface MessageGroup { + + /** + * Add a message to the internal list. This is needed to avoid hitting the + * underlying store or copying the internal list. Use with care. + */ + boolean add(Message message); + + /** + * @return internal message list, modification is allowed, but not + * recommended + */ + Collection> getUnmarked(); + + /** + * @return internal message list, modification is allowed, but not + * recommended + */ + Collection> getMarked(); + + /** + * @return the correlation key that links these messages together according + * to a particular CorrelationStrategy + */ + Object getCorrelationKey(); + + boolean isComplete(); + + int getSequenceSize(); + + int size(); + + void mark(); + + public Message getOne(); + +} \ No newline at end of file diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/store/MessageGroupStore.java b/org.springframework.integration/src/main/java/org/springframework/integration/store/MessageGroupStore.java new file mode 100644 index 0000000000..610ae743ad --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/store/MessageGroupStore.java @@ -0,0 +1,66 @@ +/* + * Copyright 2002-2008 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 + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.store; + +import java.util.Collection; + +import org.springframework.integration.core.Message; + +/** + * Interface for storage operations on groups of messages linked by a correlation key. + * + * @author Dave Syer + * + * @since 2.0 + * + */ +public interface MessageGroupStore { + + /** + * Return all Messages currently in the MessageStore that were stored using + * {@link #addMessageToGroup(Object, Collection)} with this correlation id. + * + * @return a group of messages, empty if none exists for this key + * + * @see org.springframework.integration.core.MessageHeaders#getCorrelationId() + */ + MessageGroup getMessageGroup(Object correlationId); + + /** + * Store a message with an association to a correlation id. This can be used to group messages together instead of + * storing them just under their id. + * + * @param correlationId the correlation id to store the message under + * @param message a message + */ + void addMessageToGroup(Object correlationId, Message message); + + /** + * Persist the mark on all the messages from the group. The group is modified in the process as all its unmarked + * messages become marked. + * + * @param group a MessageGroup with no unmarked messages + */ + void mark(MessageGroup group); + + /** + * Delete all the messages from the association with this correlation id. + * + * @param correlationId the correlation id to remove + */ + void deleteMessageGroup(Object correlationId); + +} \ No newline at end of file diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/store/MessageStore.java b/org.springframework.integration/src/main/java/org/springframework/integration/store/MessageStore.java index 61982fde85..6a44fa0e9c 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/store/MessageStore.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/store/MessageStore.java @@ -16,78 +16,40 @@ package org.springframework.integration.store; -import java.util.Collection; import java.util.UUID; import org.springframework.integration.core.Message; -import org.springframework.integration.core.MessageHeaders; /** - * Strategy interface for storing and retrieving messages. The interface mimics the semantics for REST for the methods - * named after REST operations. This is helpful when mapping to a RESTful API. + * Strategy interface for storing and retrieving messages. * * @author Mark Fisher * @author Iwein Fuld * @author Dave Syer + * * @since 2.0 */ public interface MessageStore { - String PROCESSED = MessageHeaders.PREFIX + "processed"; - /** * Return the Message with the given id, or null if no Message with that id exists in the MessageStore. */ - Message get(UUID id); + Message getMessage(UUID id); /** * Put the provided Message into the MessageStore. The store may need to mutate the message internally, and if it * does then the return value can be different than the input. The id of the return value will be used as an index - * so that the {@link #get(UUID)} and {@link #delete(Object)} behave properly. Since messages are immutable, putting + * so that the {@link #getMessage(UUID)} and {@link #removeMessage(UUID)} behave properly. Since messages are immutable, putting * the same message more than once is a no-op. * * @return the message that was stored */ - Message put(Message message); + Message addMessage(Message message); /** * Remove the Message with the given id from the MessageStore, if present, and return it. If no Message with that id * is present in the store, this will return null. */ - Message delete(UUID id); - - /** - * Return all Messages currently in the MessageStore that were stored using {@link #put(Object, Message)} or - * {@link #put(Object, Collection)} with this correlation id. - * - * @see org.springframework.integration.core.MessageHeaders#getCorrelationId() - */ - Collection> list(Object correlationId); - - /** - * Store a message with an association to a correlation id. This can be used to group messages together instead of - * storing them just under their id. - * - * @param correlationId the correlation id to store the message under - * @param message a message - */ - void put(Object correlationId, Message message); - - /** - * Mark a message from the association with this correlation id. If the message was previously added using - * {@link #put(Object, Message)} then it will be removed and re-inserted with the {@value #PROCESSED} flag set in - * the headers. - * - * @param correlationId the correlation id to mark the message under - */ - Message mark(Object correlationId, UUID messageId); - - /** - * Delete all the messages from the association with this correlation id. If the messages were stored under their id - * through {@link #put(Message)} they are still accessible via {@link #get(UUID)}. - * - * @param correlationId the correlation id to delete all messages under - */ - void deleteAll(Object correlationId); + Message removeMessage(UUID id); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageGroup.java b/org.springframework.integration/src/main/java/org/springframework/integration/store/SimpleMessageGroup.java similarity index 55% rename from org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageGroup.java rename to org.springframework.integration/src/main/java/org/springframework/integration/store/SimpleMessageGroup.java index 83b311c26d..f4134b806e 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageGroup.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/store/SimpleMessageGroup.java @@ -11,18 +11,18 @@ * specific language governing permissions and limitations under the License. */ -package org.springframework.integration.aggregator; +package org.springframework.integration.store; import java.util.Collection; import java.util.HashSet; import org.springframework.integration.core.Message; -import org.springframework.integration.store.MessageStore; /** - * Represents a mutable group of correlated messages that is bound to a certain {@link MessageStore} and correlation - * key. The group will grow during its lifetime, when messages are added to it. This is not thread - * safe and should not be used for long running aggregations. + * Represents a mutable group of correlated messages that is bound to a certain + * {@link MessageStore} and correlation key. The group will grow during its + * lifetime, when messages are added to it. This is not + * thread safe and should not be used for long running aggregations. * * @author Iwein Fuld * @author Oleg Zhurakousky @@ -30,59 +30,48 @@ import org.springframework.integration.store.MessageStore; * * @since 2.0 */ -public class MessageGroup { +public class SimpleMessageGroup implements MessageGroup { private final Object correlationKey; - private final Collection> marked = new HashSet>(); + public final Collection> marked = new HashSet>(); - private final Collection> unmarked = new HashSet>(); + public final Collection> unmarked = new HashSet>(); - public MessageGroup(Object correlationKey) { + public SimpleMessageGroup(Object correlationKey) { this.correlationKey = correlationKey; } - public MessageGroup(Collection> originalMessages, Object correlationKey) { + public SimpleMessageGroup(MessageGroup template) { + this.correlationKey = template.getCorrelationKey(); + this.marked.addAll(template.getMarked()); + this.unmarked.addAll(template.getUnmarked()); + } + + public SimpleMessageGroup(Collection> originalMessages, + Object correlationKey) { this(correlationKey); for (Message message : originalMessages) { add(message); } } - /** - * Add a message to the internal list. This is needed to avoid hitting the underlying store or copying the internal - * list. Use with care. - */ public boolean add(Message message) { if (isMember(message)) { return false; } - if (message.getHeaders().containsKey(MessageStore.PROCESSED) - && message.getHeaders().get(MessageStore.PROCESSED, Boolean.class)) { - this.marked.add(message); - } else { - this.unmarked.add(message); - } + this.unmarked.add(message); return true; } - /** - * @return internal message list, modification is allowed, but not recommended - */ public Collection> getUnmarked() { return unmarked; } - /** - * @return internal message list, modification is allowed, but not recommended - */ public Collection> getMarked() { return marked; } - /** - * @return the correlation key that links these messages together according to a particular CorrelationStrategy - */ public Object getCorrelationKey() { return correlationKey; } @@ -103,26 +92,35 @@ public class MessageGroup { return getOne().getHeaders().getSequenceSize(); } - private Message getOne() { - return unmarked.isEmpty() ? (marked.isEmpty() ? null : marked.iterator().next()) : unmarked.iterator().next(); + public void mark() { + marked.addAll(unmarked); + unmarked.clear(); } public int size() { return marked.size() + unmarked.size(); } + public Message getOne() { + return unmarked.isEmpty() ? (marked.isEmpty() ? null : marked + .iterator().next()) : unmarked.iterator().next(); + } + /** - * This method determines whether messages have been added to this group that supersede the given message based on - * its sequence id. This can be helpful to avoid ending up with sequences larger than their required sequence size - * or sequences that are missing certain sequence numbers. + * This method determines whether messages have been added to this group + * that supersede the given message based on its sequence id. This can be + * helpful to avoid ending up with sequences larger than their required + * sequence size or sequences that are missing certain sequence numbers. */ private boolean isMember(Message message) { if (size() == 0) { return false; } - Integer messageSequenceNumber = message.getHeaders().getSequenceNumber(); + Integer messageSequenceNumber = message.getHeaders() + .getSequenceNumber(); if (messageSequenceNumber != null && messageSequenceNumber > 0) { - Integer messageSequenceSize = message.getHeaders().getSequenceSize(); + Integer messageSequenceSize = message.getHeaders() + .getSequenceSize(); if (!messageSequenceSize.equals(getSequenceSize()) || containsSequenceNumber(unmarked, messageSequenceNumber) || containsSequenceNumber(marked, messageSequenceNumber)) { @@ -132,9 +130,11 @@ public class MessageGroup { return false; } - private boolean containsSequenceNumber(Collection> messages, Integer messageSequenceNumber) { + private boolean containsSequenceNumber(Collection> messages, + Integer messageSequenceNumber) { for (Message member : messages) { - Integer memberSequenceNumber = member.getHeaders().getSequenceNumber(); + Integer memberSequenceNumber = member.getHeaders() + .getSequenceNumber(); if (messageSequenceNumber.equals(memberSequenceNumber)) { return true; } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/store/SimpleMessageStore.java b/org.springframework.integration/src/main/java/org/springframework/integration/store/SimpleMessageStore.java index 43eaf5f334..224cf97bda 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/store/SimpleMessageStore.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/store/SimpleMessageStore.java @@ -16,43 +16,40 @@ package org.springframework.integration.store; -import java.util.Collection; -import java.util.Collections; -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.core.Message; import org.springframework.integration.core.MessagingException; -import org.springframework.integration.message.MessageBuilder; import org.springframework.integration.util.UpperBound; import org.springframework.util.Assert; /** - * Map-based implementation of {@link MessageStore} that enforces a maximum - * capacity. + * Map-based implementation of {@link MessageStore} and {@link MessageGroupStore}. Enforces a maximum capacity for the + * store. * * @author Iwein Fuld * @author Mark Fisher + * @author Dave Syer + * * @since 2.0 */ -public class SimpleMessageStore implements MessageStore { +public class SimpleMessageStore implements MessageStore, MessageGroupStore { private final ConcurrentMap> idToMessage; - private final ConcurrentMap>> correlationToMessage; + private final ConcurrentMap correlationToMessageGroup; private final UpperBound upperBound; /** - * Creates a SimpleMessageStore with a maximum size limited by the given - * capacity, or unlimited size if the given capacity is less than 1. + * Creates a SimpleMessageStore with a maximum size limited by the given capacity, or unlimited size if the given + * capacity is less than 1. */ public SimpleMessageStore(int capacity) { this.idToMessage = new ConcurrentHashMap>(); - this.correlationToMessage = new ConcurrentHashMap>>(); + this.correlationToMessageGroup = new ConcurrentHashMap(); this.upperBound = new UpperBound(capacity); } @@ -64,23 +61,23 @@ public class SimpleMessageStore implements MessageStore { } @SuppressWarnings("unchecked") - public Message put(Message message) { + public Message addMessage(Message message) { if (!upperBound.tryAcquire(0)) { throw new MessagingException(this.getClass().getSimpleName() + " was out of capacity at, try constructing it with a larger capacity."); } Object correlationId = message.getHeaders().getCorrelationId(); - if (correlationId!=null) { - getMessagesInternal(correlationId).add(message); + if (correlationId != null) { + getMessageGroupInternal(correlationId).add(message); } return (Message) this.idToMessage.put(message.getHeaders().getId(), message); } - public Message get(UUID key) { + public Message getMessage(UUID key) { return (key != null) ? this.idToMessage.get(key) : null; } - public Message delete(UUID key) { + public Message removeMessage(UUID key) { if (key != null) { upperBound.release(); return this.idToMessage.remove(key); @@ -93,46 +90,35 @@ public class SimpleMessageStore implements MessageStore { return this.idToMessage.size(); } - public Collection> list(Object correlationId) { + public MessageGroup getMessageGroup(Object correlationId) { Assert.notNull(correlationId, "'correlationKey' must not be null"); - Collection> collection = correlationToMessage.get(correlationId); - if (collection==null) { - return Collections.emptySet(); + MessageGroup collection = correlationToMessageGroup.get(correlationId); + if (collection == null) { + return new SimpleMessageGroup(correlationId); } - return new HashSet>(collection); + return new SimpleMessageGroup(collection); } - public void put(Object correlationId, Message message) { - getMessagesInternal(correlationId).add(message); + public void addMessageToGroup(Object correlationId, Message message) { + getMessageGroupInternal(correlationId).add(message); + } + + public void mark(MessageGroup group) { + Object correlationId = group.getCorrelationKey(); + MessageGroup internal = getMessageGroupInternal(correlationId); + internal.mark(); + group.mark(); } - public Message mark(Object correlationId, UUID messageId) { - if (!correlationToMessage.containsKey(correlationId)) { - return null; - } - Collection> messages = getMessagesInternal(correlationId); - Message result = null; - for (Iterator> iterator = messages.iterator(); iterator.hasNext();) { - Message message = (Message) iterator.next(); - if (message.getHeaders().getId().equals(messageId)) { - iterator.remove(); - result = MessageBuilder.fromMessage(message).setHeader(PROCESSED, true).build(); - } - } - messages.add(result); - return result; + public void deleteMessageGroup(Object correlationId) { + correlationToMessageGroup.remove(correlationId); } - public void deleteAll(Object correlationId) { - correlationToMessage.remove(correlationId); - } - - private Collection> getMessagesInternal(Object correlationId) { - if (!correlationToMessage.containsKey(correlationId)) { - correlationToMessage.putIfAbsent(correlationId, new HashSet>()); + private MessageGroup getMessageGroupInternal(Object correlationId) { + if (!correlationToMessageGroup.containsKey(correlationId)) { + correlationToMessageGroup.putIfAbsent(correlationId, new SimpleMessageGroup(correlationId)); } - Collection> collection = correlationToMessage.get(correlationId); - return collection; + return correlationToMessageGroup.get(correlationId); } } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/transformer/ClaimCheckTransformer.java b/org.springframework.integration/src/main/java/org/springframework/integration/transformer/ClaimCheckTransformer.java index 0d9beff893..a2fb3f804d 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/transformer/ClaimCheckTransformer.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/transformer/ClaimCheckTransformer.java @@ -59,13 +59,13 @@ public class ClaimCheckTransformer extends AbstractTransformer { private Message retrieveMessage(UUID id) { - Message result = this.messageStore.get(id); + Message result = this.messageStore.getMessage(id); Assert.notNull(result, "unable to locate Message for claim check ID: " + id); return result; } private void storeMessage(Message message) { - this.messageStore.put(message); + this.messageStore.addMessage(message); } } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/AggregatingMessageGroupProcessorHeaderTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/AggregatingMessageGroupProcessorHeaderTests.java index c8f7532e5d..592cdfb513 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/AggregatingMessageGroupProcessorHeaderTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/AggregatingMessageGroupProcessorHeaderTests.java @@ -31,6 +31,8 @@ import org.springframework.integration.channel.MessageChannelTemplate; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.core.Message; import org.springframework.integration.message.MessageBuilder; +import org.springframework.integration.store.MessageGroup; +import org.springframework.integration.store.SimpleMessageGroup; /** * @author Mark Fisher @@ -104,7 +106,7 @@ public class AggregatingMessageGroupProcessorHeaderTests { headers.put("k2", new Integer(2)); Message message = correlatedMessage(1, 1, 1, headers); List> messages = Collections.>singletonList(message); - MessageGroup group = new MessageGroup(messages, 1); + MessageGroup group = new SimpleMessageGroup(messages, 1); processor.processAndSend(group, channelTemplate, outputChannel); Message result = outputChannel.receive(0); assertNotNull(result); @@ -119,7 +121,7 @@ public class AggregatingMessageGroupProcessorHeaderTests { Message message1 = correlatedMessage(1, 2, 1, headers); Message message2 = correlatedMessage(1, 2, 2, headers); List> messages = Arrays.>asList(message1, message2); - MessageGroup group = new MessageGroup(messages, 1); + MessageGroup group = new SimpleMessageGroup(messages, 1); processor.processAndSend(group, channelTemplate, outputChannel); Message result = outputChannel.receive(0); assertNotNull(result); @@ -137,7 +139,7 @@ public class AggregatingMessageGroupProcessorHeaderTests { headers2.put("k2", new Integer(123)); Message message2 = correlatedMessage(1, 2, 2, headers2); List> messages = Arrays.>asList(message1, message2); - MessageGroup group = new MessageGroup(messages, 1); + MessageGroup group = new SimpleMessageGroup(messages, 1); processor.processAndSend(group, channelTemplate, outputChannel); Message result = outputChannel.receive(0); assertNotNull(result); @@ -167,7 +169,7 @@ public class AggregatingMessageGroupProcessorHeaderTests { headers3.put("conflictBetween2And3", "valueFor3"); Message message3 = correlatedMessage(1, 3, 3, headers3); List> messages = Arrays.>asList(message1, message2, message3); - MessageGroup group = new MessageGroup(messages, 1); + MessageGroup group = new SimpleMessageGroup(messages, 1); processor.processAndSend(group, channelTemplate, outputChannel); Message result = outputChannel.receive(0); assertNotNull(result); @@ -195,7 +197,7 @@ public class AggregatingMessageGroupProcessorHeaderTests { headers3.put("common", "valueForAll"); Message message3 = correlatedMessage(1, 3, 3, headers3); List> messages = Arrays.>asList(message1, message2, message3); - MessageGroup group = new MessageGroup(messages, 1); + MessageGroup group = new SimpleMessageGroup(messages, 1); processor.processAndSend(group, channelTemplate, outputChannel); Message result = outputChannel.receive(0); assertNotNull(result); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java index ba86bd6da8..bcb9f6f1e4 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java @@ -35,6 +35,7 @@ import org.springframework.integration.core.MessageChannel; import org.springframework.integration.core.MessageHeaders; import org.springframework.integration.message.MessageBuilder; import org.springframework.integration.message.MessageHandlingException; +import org.springframework.integration.store.MessageGroup; import org.springframework.integration.store.SimpleMessageStore; /** diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/ConcurrentAggregatorTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/ConcurrentAggregatorTests.java index 9eb9b3ff1c..c36ade6438 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/ConcurrentAggregatorTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/ConcurrentAggregatorTests.java @@ -38,6 +38,7 @@ import org.springframework.integration.core.MessageHeaders; import org.springframework.integration.message.MessageBuilder; import org.springframework.integration.message.MessageHandler; import org.springframework.integration.message.MessageHandlingException; +import org.springframework.integration.store.MessageGroup; import org.springframework.integration.store.SimpleMessageStore; /** diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerIntegrationTest.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerIntegrationTest.java index 98e9385828..e0759119f0 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerIntegrationTest.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerIntegrationTest.java @@ -25,12 +25,12 @@ import org.junit.Test; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.message.MessageBuilder; -import org.springframework.integration.store.MessageStore; +import org.springframework.integration.store.MessageGroupStore; import org.springframework.integration.store.SimpleMessageStore; public class CorrelatingMessageHandlerIntegrationTest { - private MessageStore store = new SimpleMessageStore(100); + private MessageGroupStore store = new SimpleMessageStore(100); private MessageChannel outputChannel = mock(MessageChannel.class); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java index 03f2fbdcf5..4e6954e414 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java @@ -40,6 +40,7 @@ import org.springframework.integration.channel.MessageChannelTemplate; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.message.MessageBuilder; +import org.springframework.integration.store.SimpleMessageGroup; import org.springframework.integration.store.SimpleMessageStore; import org.springframework.test.util.ReflectionTestUtils; @@ -68,7 +69,7 @@ public class CorrelatingMessageHandlerTests { handler = new CorrelatingMessageHandler(processor, new SimpleMessageStore(), correlationStrategy, ReleaseStrategy); handler.setOutputChannel(outputChannel); - doAnswer(new DoesNothing()).when(processor).processAndSend(isA(MessageGroup.class), + doAnswer(new DoesNothing()).when(processor).processAndSend(isA(SimpleMessageGroup.class), isA(MessageChannelTemplate.class), eq(outputChannel)); } @@ -91,7 +92,7 @@ public class CorrelatingMessageHandlerTests { verify(correlationStrategy).getCorrelationKey(message1); verify(correlationStrategy).getCorrelationKey(message2); - verify(processor).processAndSend(isA(MessageGroup.class), isA(MessageChannelTemplate.class), eq(outputChannel)); + verify(processor).processAndSend(isA(SimpleMessageGroup.class), isA(MessageChannelTemplate.class), eq(outputChannel)); } private void verifyLocks(CorrelatingMessageHandler handler, int lockCount) { diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessorTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessorTests.java index cbf0b65d44..9e04ae433f 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessorTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessorTests.java @@ -30,6 +30,7 @@ import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.endpoint.EventDrivenConsumer; import org.springframework.integration.message.MessageBuilder; +import org.springframework.integration.store.MessageGroup; @RunWith(MockitoJUnitRunner.class) public class MethodInvokingMessageGroupProcessorTests { diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/ReleaseStrategyAdapterTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/ReleaseStrategyAdapterTests.java index 6747a32329..653aef0da0 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/ReleaseStrategyAdapterTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/ReleaseStrategyAdapterTests.java @@ -25,6 +25,8 @@ import org.junit.Before; import org.junit.Test; import org.springframework.integration.core.Message; import org.springframework.integration.message.GenericMessage; +import org.springframework.integration.store.MessageGroup; +import org.springframework.integration.store.SimpleMessageGroup; /** * @author Marius Bogoevici @@ -155,7 +157,7 @@ public class ReleaseStrategyAdapterTests { if (size > 2) { messages.add(new GenericMessage("789")); } - return new MessageGroup(messages, "ABC"); + return new SimpleMessageGroup(messages, "ABC"); } @SuppressWarnings("unused") diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java index 1251675726..845188ec33 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java @@ -19,7 +19,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 static org.junit.Assert.assertTrue; import java.util.ArrayList; import java.util.Arrays; @@ -31,12 +30,13 @@ import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.message.MessageBuilder; -import org.springframework.integration.store.MessageStore; +import org.springframework.integration.store.MessageGroupStore; import org.springframework.integration.store.SimpleMessageStore; /** * @author Marius Bogoevici * @author Alex Peters + * @author Dave Syer */ public class ResequencerTests { @@ -44,7 +44,7 @@ public class ResequencerTests { private Resequencer processor = new Resequencer(); - private MessageStore store = new SimpleMessageStore(); + private MessageGroupStore store = new SimpleMessageStore(); @Before public void configureResequencer() { @@ -230,7 +230,7 @@ public class ResequencerTests { String correlationId = "ABC"; Message message1 = createMessage("123", correlationId, 1, 1, replyChannel); resequencer.handleMessage(message1); - assertTrue(store.list(correlationId).isEmpty()); + assertEquals(0, store.getMessageGroup(correlationId).size()); } private static Message createMessage(String payload, Object correlationId, int sequenceSize, int sequenceNumber, diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/SequenceSizeReleaseStrategyTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/SequenceSizeReleaseStrategyTests.java index 6becb100af..690a158adf 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/SequenceSizeReleaseStrategyTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/SequenceSizeReleaseStrategyTests.java @@ -22,6 +22,8 @@ import static org.junit.Assert.assertTrue; import org.junit.Test; import org.springframework.integration.core.Message; import org.springframework.integration.message.MessageBuilder; +import org.springframework.integration.store.MessageGroup; +import org.springframework.integration.store.SimpleMessageGroup; /** * @author Mark Fisher @@ -32,7 +34,7 @@ public class SequenceSizeReleaseStrategyTests { public void testIncompleteList() { Message message = MessageBuilder.withPayload("test1") .setSequenceSize(2).build(); - MessageGroup messages = new MessageGroup("FOO"); + MessageGroup messages = new SimpleMessageGroup("FOO"); messages.add(message); SequenceSizeReleaseStrategy ReleaseStrategy = new SequenceSizeReleaseStrategy(); assertFalse(ReleaseStrategy.canRelease(messages)); @@ -44,7 +46,7 @@ public class SequenceSizeReleaseStrategyTests { .setSequenceSize(2).build(); Message message2 = MessageBuilder.withPayload("test2") .setSequenceSize(2).build(); - MessageGroup messages = new MessageGroup("FOO"); + MessageGroup messages = new SimpleMessageGroup("FOO"); messages.add(message1); messages.add(message2); SequenceSizeReleaseStrategy ReleaseStrategy = new SequenceSizeReleaseStrategy(); @@ -54,7 +56,7 @@ public class SequenceSizeReleaseStrategyTests { @Test public void testEmptyList() { SequenceSizeReleaseStrategy ReleaseStrategy = new SequenceSizeReleaseStrategy(); - assertTrue(ReleaseStrategy.canRelease(new MessageGroup("FOO"))); + assertTrue(ReleaseStrategy.canRelease(new SimpleMessageGroup("FOO"))); } } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorWithCorrelationStrategyTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorWithCorrelationStrategyTests.java index 50b0380ce0..fe20a10ad6 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorWithCorrelationStrategyTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorWithCorrelationStrategyTests.java @@ -21,7 +21,6 @@ import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; -import org.springframework.integration.aggregator.MessageGroup; import org.springframework.integration.aggregator.ReleaseStrategy; import org.springframework.integration.aggregator.CorrelationStrategy; import org.springframework.integration.annotation.Aggregator; @@ -29,6 +28,7 @@ import org.springframework.integration.channel.PollableChannel; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.message.MessageBuilder; +import org.springframework.integration.store.MessageGroup; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/TestReleaseStrategy.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/TestReleaseStrategy.java index 01e40d80b7..a696dd353c 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/TestReleaseStrategy.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/TestReleaseStrategy.java @@ -17,8 +17,8 @@ package org.springframework.integration.config; -import org.springframework.integration.aggregator.MessageGroup; import org.springframework.integration.aggregator.ReleaseStrategy; +import org.springframework.integration.store.MessageGroup; /** * @author Marius Bogoevici diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/MessageGroupTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/store/SimpleMessageGroupTests.java similarity index 83% rename from org.springframework.integration/src/test/java/org/springframework/integration/aggregator/MessageGroupTests.java rename to org.springframework.integration/src/test/java/org/springframework/integration/store/SimpleMessageGroupTests.java index 8531758c6e..ef6a8b0009 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/MessageGroupTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/store/SimpleMessageGroupTests.java @@ -1,4 +1,4 @@ -package org.springframework.integration.aggregator; +package org.springframework.integration.store; import static org.hamcrest.CoreMatchers.is; import static org.junit.Assert.assertThat; @@ -9,13 +9,14 @@ import org.junit.Before; import org.junit.Test; import org.springframework.integration.core.Message; import org.springframework.integration.message.MessageBuilder; +import org.springframework.integration.store.SimpleMessageGroup; /** * @author Iwein Fuld * @author Oleg Zhurakousky * @author Dave Syer */ -public class MessageGroupTests { +public class SimpleMessageGroupTests { private Object key = new Object(); @@ -23,7 +24,7 @@ public class MessageGroupTests { @Before public void buildMessageGroup() { - group = new MessageGroup(Collections.> emptyList(), key); + group = new SimpleMessageGroup(Collections.> emptyList(), key); } @Test diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/store/SimpleMessageStoreTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/store/SimpleMessageStoreTests.java index d5e12e7745..5fcf49aeac 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/store/SimpleMessageStoreTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/store/SimpleMessageStoreTests.java @@ -18,6 +18,7 @@ package org.springframework.integration.store; import static org.hamcrest.CoreMatchers.is; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotSame; import static org.junit.Assert.assertThat; import org.junit.Test; @@ -36,8 +37,8 @@ public class SimpleMessageStoreTests { public void shouldRetainMessage() { SimpleMessageStore store = new SimpleMessageStore(); Message testMessage1 = MessageBuilder.withPayload("foo").build(); - store.put(testMessage1); - assertThat((Message) store.get(testMessage1.getHeaders().getId()), is(testMessage1)); + store.addMessage(testMessage1); + assertThat((Message) store.getMessage(testMessage1.getHeaders().getId()), is(testMessage1)); } @Test(expected = MessagingException.class) @@ -45,8 +46,8 @@ public class SimpleMessageStoreTests { SimpleMessageStore store = new SimpleMessageStore(1); Message testMessage1 = MessageBuilder.withPayload("foo").build(); Message testMessage2 = MessageBuilder.withPayload("bar").build(); - store.put(testMessage1); - store.put(testMessage2); + store.addMessage(testMessage1); + store.addMessage(testMessage2); } @Test @@ -54,16 +55,24 @@ public class SimpleMessageStoreTests { SimpleMessageStore store = new SimpleMessageStore(2); Message testMessage1 = MessageBuilder.withPayload("foo").build(); Message testMessage2 = MessageBuilder.withPayload("bar").build(); - store.put(testMessage1); - store.put(testMessage2); + store.addMessage(testMessage1); + store.addMessage(testMessage2); } @Test public void shouldListByCorrelation() throws Exception { SimpleMessageStore store = new SimpleMessageStore(); Message testMessage1 = MessageBuilder.withPayload("foo").build(); - store.put("bar", testMessage1); - assertEquals(1, store.list("bar").size()); + store.addMessageToGroup("bar", testMessage1); + assertEquals(1, store.getMessageGroup("bar").size()); + } + + @Test + public void shouldCopyMessageGroup() throws Exception { + SimpleMessageStore store = new SimpleMessageStore(); + Message testMessage1 = MessageBuilder.withPayload("foo").build(); + store.addMessageToGroup("bar", testMessage1); + assertNotSame(store.getMessageGroup("bar"), store.getMessageGroup("bar")); } } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/transformer/ClaimCheckTransformerTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/transformer/ClaimCheckTransformerTests.java index 46a7046d61..e1e2d0773f 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/transformer/ClaimCheckTransformerTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/transformer/ClaimCheckTransformerTests.java @@ -47,7 +47,7 @@ public class ClaimCheckTransformerTests { MessageStore store = new SimpleMessageStore(10); Message message = MessageBuilder.withPayload("test").build(); UUID storedId = message.getHeaders().getId(); - store.put(message); + store.addMessage(message); ClaimCheckTransformer transformer = new ClaimCheckTransformer(store); Message input = MessageBuilder.withPayload(storedId).build(); Message output = transformer.transform(input); diff --git a/spring-integration-parent/.classpath b/spring-integration-parent/.classpath index d0bec0f761..a225f00dbd 100644 --- a/spring-integration-parent/.classpath +++ b/spring-integration-parent/.classpath @@ -1,6 +1,6 @@ - +