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 b13ba286be..698bd902ba 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 @@ -1,45 +1,70 @@ package org.springframework.integration.store; -import java.util.Collection; - import org.springframework.integration.core.Message; +import java.util.Collection; + +/** + * A group of messages that are correlated with each other and should be processed in the same context. The group is + * divided into marked and unmarked messages. The marked messages are typically already processed, the unmarked messages + * are to be processed in the future. + *

+ * The message group allows implementations to be mutable, but this behavior is optional. Implementations should take + * care to document their thread safety and mutability. + */ 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. + * underlying store or copying the internal list. */ boolean add(Message message); /** - * @return internal message list, modification is allowed, but not - * recommended + * @return unmarked messages in the group at time of the invocation */ Collection> getUnmarked(); /** - * @return internal message list, modification is allowed, but not - * recommended + * @return marked messages in the group at the time of the invocation */ Collection> getMarked(); /** - * @return the correlation key that links these messages together according + * @return the correlation key that links these messages together, typically according * to a particular CorrelationStrategy */ Object getCorrelationKey(); + /** + * @return true if the group is complete (i.e. no more messages are expected to be added) + */ boolean isComplete(); + /** + * @return the size of the sequence expected 0 if unknown + */ int getSequenceSize(); + /** + * @return the total number of messages (marked and unmarked) in this group + */ int size(); + /** + * Mark all unmarked messages in the group. A MessageGroupProcessor typically invokes this method after + * processing all unmarked messages. + */ void mark(); + /** + * @return a single message from the group + */ Message getOne(); - + + /** + * @return the timestamp (milliseconds since epoch) associated with the creation of this group + */ long getTimestamp(); } \ No newline at end of file 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 f59f48e82c..1a2f19f096 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 @@ -22,7 +22,8 @@ import java.util.concurrent.LinkedBlockingQueue; /** * 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. + * key. The group will grow during its lifetime, when messages are added to it. This MessageGroup is thread + * safe. * * @author Iwein Fuld * @author Oleg Zhurakousky 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 73837dd87a..b35c207ed6 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.core.Message; +import org.springframework.integration.core.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.core.Message; -import org.springframework.integration.core.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. @@ -38,7 +38,7 @@ public class SimpleMessageStore extends AbstractMessageGroupStore implements Mes private final ConcurrentMap> idToMessage; - final ConcurrentMap correlationToMessageGroup; + private final ConcurrentMap correlationToMessageGroup; private final UpperBound individualUpperBound;