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

INT-1940 added correlationLocksMonitor to CorrelatingMessageHandler to monitor 'locks' access

Browse Source
This commit is contained in:
Oleg Zhurakousky
2011-06-23 15:36:47 -04:00
parent 8737cf85e3
commit d8fc7f939d
1 changed files with 11 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
spring-integration-core/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java
View File

@@ -47,6 +47,7 @@ import java.util.concurrent.ConcurrentMap;
*
* @author Iwein Fuld
* @author Dave Syer
* @author Oleg Zhurakousky
* @since 2.0
*/
public class CorrelatingMessageHandler extends AbstractMessageHandler implements MessageProducer {
@@ -71,6 +72,8 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
private volatile MessageChannel discardChannel = new NullChannel();
private boolean sendPartialResultOnExpiry = false;
private final Object correlationLocksMonitor = new Object();
private final ConcurrentMap<Object, Object> locks = new ConcurrentHashMap<Object, Object>();
@@ -166,7 +169,7 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
// TODO: INT-1117 - make the lock global?
Object lock = getLock(correlationKey);
Assert.notNull(lock, "Based on correlationKey '" + correlationKey + "' the returned lock is null"); // see comments in INT-1940
synchronized (lock) {
MessageGroup group = messageStore.getMessageGroup(correlationKey);
if (group.canAdd(message)) {
@@ -243,8 +246,10 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
}
private Object getLock(Object correlationKey) {
locks.putIfAbsent(correlationKey, correlationKey);
return locks.get(correlationKey);
synchronized(correlationLocksMonitor){
locks.putIfAbsent(correlationKey, correlationKey);
return locks.get(correlationKey);
}
}
private void mark(MessageGroup group) {
@@ -262,7 +267,9 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
private void remove(MessageGroup group) {
Object correlationKey = group.getGroupId();
messageStore.removeMessageGroup(correlationKey);
locks.remove(correlationKey);
synchronized(correlationLocksMonitor){
locks.remove(correlationKey);
}
}
private MessageGroup store(Object correlationKey, Message<?> message) {