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

INT-1054: Fix bug in SimpleMesssageStore; INT-1112: Tweak method names in MessageGroupStore

Browse Source
This commit is contained in:
David Syer
2010-05-04 05:56:42 +00:00
parent f697c2f386
commit 5dfa2d48f3
5 changed files with 21 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java
View File

@@ -249,12 +249,12 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
}
private void mark(MessageGroup group) {
store.mark(group);
store.markMessageGroup(group);
}
private void remove(MessageGroup group) {
Object correlationKey = group.getCorrelationKey();
store.deleteMessageGroup(correlationKey);
store.removeMessageGroup(correlationKey);
locks.remove(correlationKey);
}

18
org.springframework.integration/src/main/java/org/springframework/integration/store/MessageGroupStore.java
View File

@@ -34,19 +34,17 @@ public interface MessageGroupStore {
* {@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);
MessageGroup getMessageGroup(Object correlationKey);
/**
* Store a message with an association to a correlation id. This can be used to group messages together instead of
* Store a message with an association to a correlation key. 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 correlationKey the correlation id to store the message under
* @param message a message
*/
void addMessageToGroup(Object correlationId, Message<?> message);
void addMessageToGroup(Object correlationKey, Message<?> message);
/**
* Persist the mark on all the messages from the group. The group is modified in the process as all its unmarked
@@ -54,13 +52,13 @@ public interface MessageGroupStore {
*
* @param group a MessageGroup with no unmarked messages
*/
void mark(MessageGroup group);
void markMessageGroup(MessageGroup group);
/**
* Delete all the messages from the association with this correlation id.
* Remove the message group with this correlation key.
*
* @param correlationId the correlation id to remove
* @param correlationKey the correlation id to remove
*/
void deleteMessageGroup(Object correlationId);
void removeMessageGroup(Object correlationKey);
}

16
org.springframework.integration/src/main/java/org/springframework/integration/store/SimpleMessageStore.java
View File

@@ -60,17 +60,13 @@ public class SimpleMessageStore implements MessageStore, MessageGroupStore {
this(0);
}
@SuppressWarnings("unchecked")
public <T> Message<T> addMessage(Message<T> 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) {
getMessageGroupInternal(correlationId).add(message);
}
return (Message<T>) this.idToMessage.put(message.getHeaders().getId(), message);
this.idToMessage.put(message.getHeaders().getId(), message);
return message;
}
public Message<?> getMessage(UUID key) {
@@ -86,10 +82,6 @@ public class SimpleMessageStore implements MessageStore, MessageGroupStore {
return null;
}
public int size() {
return this.idToMessage.size();
}
public MessageGroup getMessageGroup(Object correlationId) {
Assert.notNull(correlationId, "'correlationKey' must not be null");
MessageGroup collection = correlationToMessageGroup.get(correlationId);
@@ -103,14 +95,14 @@ public class SimpleMessageStore implements MessageStore, MessageGroupStore {
getMessageGroupInternal(correlationId).add(message);
}
public void mark(MessageGroup group) {
public void markMessageGroup(MessageGroup group) {
Object correlationId = group.getCorrelationKey();
MessageGroup internal = getMessageGroupInternal(correlationId);
internal.mark();
group.mark();
}
public void deleteMessageGroup(Object correlationId) {
public void removeMessageGroup(Object correlationId) {
correlationToMessageGroup.remove(correlationId);
}