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

javadoc and reduced visibility of add(Message m)

Browse Source
This commit is contained in:
Iwein Fuld
2010-04-06 16:45:09 +00:00
parent 8ec441fc2f
commit c98e17ee35
1 changed files with 31 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

37
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageGroup.java
View File

@@ -5,7 +5,8 @@ import org.springframework.integration.core.Message;
import java.util.*;
/**
* Represents an mutable group of correlated messages that is bound to a certain {@link org.springframework.integration.store.MessageStore}
* Represents an mutable group of correlated messages that is bound to a certain
* {@link org.springframework.integration.store.MessageStore}
* and correlation key. The group will grow during its lifetime, when messages are <code>add</code>ed to it. <strong>
* This is not thread safe and should not be used for long running aggregations</strong>.
* <p/>
@@ -16,7 +17,6 @@ import java.util.*;
* group is completed.
*
*
*
* @author Iwein Fuld
* @author Oleg Zhurakousky
*/
@@ -27,14 +27,19 @@ public class MessageGroup {
private final List<MessageGroupListener> listeners;
public MessageGroup(Collection<? extends Message<?>> originalMessages, CompletionStrategy completionStrategy, Object correlationKey,
MessageGroupListener... listeners) {
public MessageGroup(Collection<? extends Message<?>> originalMessages, CompletionStrategy completionStrategy,
Object correlationKey, MessageGroupListener... listeners) {
this.completionStrategy = completionStrategy;
this.correlationKey = correlationKey;
this.messages.addAll(originalMessages);
this.listeners = Collections.unmodifiableList(Arrays.asList(listeners));
}
/**
* 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.
*/
public boolean hasNoMessageSuperseding(Message<?> message) {
Integer messageSequenceNumber = message.getHeaders().getSequenceNumber();
if (messageSequenceNumber > 0){
@@ -43,7 +48,7 @@ public class MessageGroup {
}
for (Message<?> member : messages) {
Integer memberSequenceNumber = member.getHeaders().getSequenceNumber();
if (memberSequenceNumber == messageSequenceNumber) {
if (memberSequenceNumber.equals(messageSequenceNumber)) {
return false;
}
}
@@ -51,28 +56,48 @@ public class MessageGroup {
return true;
}
public void add(Message<?> 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.
*/
protected void add(Message<?> message) {
messages.add(message);
}
/**
* {@inheritDoc}
*/
public boolean isComplete() {
return completionStrategy.isComplete(messages);
}
/**
* @return internal message list, modification is allowed, but not recommended
*/
public List<Message<?>> getMessages() {
return messages;
}
/**
* @return the correlation key that links these messages together according to a particular CorrelationStrategy
*/
public Object getCorrelationKey() {
return correlationKey;
}
/**
* Call this method to sign off on processing of certain messages e.g. from a MessageProcessor. Typically this will
* remove these messages from the processing backlog.
*/
public void onProcessingOf(Message... messages) {
for (MessageGroupListener listener : listeners) {
listener.onProcessingOf(messages);
}
}
/**
* Call this method to signal the completion of the processing of an entire group
*/
public void onCompletion() {
for (MessageGroupListener listener : listeners) {
listener.onCompletionOf(correlationKey);