diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java index 8d2aaf671b..7746ef21b3 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java @@ -1,11 +1,11 @@ /* * Copyright 2002-2012 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. @@ -44,10 +44,10 @@ import org.springframework.util.CollectionUtils; /** * Abstract Message handler that holds a buffer of correlated messages in a * {@link MessageStore}. This class takes care of correlated groups of messages - * that can be completed in batches. It is useful for custom implementation of MessageHandlers that require correlation + * that can be completed in batches. It is useful for custom implementation of MessageHandlers that require correlation * and is used as a base class for Aggregator - {@link AggregatingMessageHandler} and * Resequencer - {@link ResequencingMessageHandler}, - * or custom implementations requiring correlation. + * or custom implementations requiring correlation. *

* To customize this handler inject {@link CorrelationStrategy}, * {@link ReleaseStrategy}, and {@link MessageGroupProcessor} implementations as @@ -60,6 +60,7 @@ import org.springframework.util.CollectionUtils; * @author Iwein Fuld * @author Dave Syer * @author Oleg Zhurakousky + * @author Gary Russell * @since 2.0 */ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageHandler implements MessageProducer { @@ -83,7 +84,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH private volatile MessageChannel discardChannel = new NullChannel(); private boolean sendPartialResultOnExpiry = false; - + private volatile boolean sequenceAware = false; private volatile LockRegistry lockRegistry = new DefaultLockRegistry(); @@ -151,6 +152,11 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH if (beanFactory != null) { this.messagingTemplate.setBeanFactory(beanFactory); } + /* + * Disallow any further changes to the lock registry + * (checked in the setter). + */ + this.lockRegistrySet = true; } public void setDiscardChannel(MessageChannel discardChannel) { @@ -176,7 +182,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH public String getComponentType() { return "aggregator"; } - + protected MessageGroupStore getMessageStore() { return messageStore; } @@ -205,7 +211,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH logger.trace("Adding message to group [ " + messageGroup + "]"); } messageGroup = this.store(correlationKey, message); - + if (releaseStrategy.canRelease(messageGroup)) { Collection> completedMessages = null; try { @@ -213,11 +219,11 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH } finally { // Always clean up even if there was an exception - // processing messages + // processing messages this.afterRelease(messageGroup, completedMessages); } - } - } + } + } else { discardChannel.send(message); } @@ -228,7 +234,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH } /** - * Allows you to provide additional logic that needs to be performed after the MessageGroup was released. + * Allows you to provide additional logic that needs to be performed after the MessageGroup was released. * @param group * @param completedMessages */ @@ -273,11 +279,11 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH } protected int findLastReleasedSequenceNumber(Object groupId, Collection> partialSequence){ - List> sorted = new ArrayList>((Collection>)partialSequence); + List> sorted = new ArrayList>(partialSequence); Collections.sort(sorted, new SequenceNumberComparator()); - + Message lastReleasedMessage = sorted.get(partialSequence.size()-1); - + return lastReleasedMessage.getHeaders().getSequenceNumber(); } @@ -319,7 +325,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH if (logger.isDebugEnabled()) { logger.debug("Completing group with correlationKey [" + correlationKey + "]"); } - + Object result = outputProcessor.processMessageGroup(group); Collection> partialSequence = null; if (result instanceof Collection) { @@ -329,7 +335,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH this.sendReplies(result, message); return partialSequence; } - + private void verifyResultCollectionConsistsOfMessages(Collection elements){ Class commonElementType = CollectionUtils.findCommonElementType(elements); Assert.isAssignable(Message.class, commonElementType, "The expected collection of Messages contains non-Message element: " + commonElementType); @@ -393,6 +399,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH * 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. */ + @Override public boolean canAdd(Message message) { if (this.size() == 0) { return true; 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 3588908c25..a098e1646f 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 @@ -1,11 +1,11 @@ /* - * Copyright 2002-2011 the original author or authors. - * + * Copyright 2002-2012 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. @@ -19,9 +19,12 @@ import java.util.Iterator; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.locks.Lock; import org.springframework.integration.Message; import org.springframework.integration.MessagingException; +import org.springframework.integration.util.DefaultLockRegistry; +import org.springframework.integration.util.LockRegistry; import org.springframework.integration.util.UpperBound; import org.springframework.jmx.export.annotation.ManagedAttribute; import org.springframework.jmx.export.annotation.ManagedResource; @@ -31,27 +34,30 @@ import org.springframework.util.CollectionUtils; /** * Map-based in-memory implementation of {@link MessageStore} and {@link MessageGroupStore}. Enforces a maximum capacity for the * store. - * + * * @author Iwein Fuld * @author Mark Fisher * @author Dave Syer * @author Oleg Zhurakousky - * + * @author Gary Russell + * * @since 2.0 */ @ManagedResource public class SimpleMessageStore extends AbstractMessageGroupStore implements MessageStore, MessageGroupStore { - - private final ConcurrentMap locks = new ConcurrentHashMap(); + + private volatile LockRegistry lockRegistry; private final ConcurrentMap> idToMessage; private final ConcurrentMap groupIdToMessageGroup; - + private final UpperBound individualUpperBound; private final UpperBound groupUpperBound; - + + private volatile boolean isUsed; + /** * Creates a SimpleMessageStore with a maximum size limited by the given capacity, or unlimited size if the given * capacity is less than 1. The capacities are applied independently to messages stored via @@ -60,10 +66,21 @@ public class SimpleMessageStore extends AbstractMessageGroupStore implements Mes * store another will result in an exception. */ public SimpleMessageStore(int individualCapacity, int groupCapacity) { + this(individualCapacity, groupCapacity, new DefaultLockRegistry()); + } + + /** + * See {@link #SimpleMessageStore(int, int)}. + * Also allows the provision of a custom {@link LockRegistry} + * rather than using the default. + */ + public SimpleMessageStore(int individualCapacity, int groupCapacity, LockRegistry lockRegistry) { + Assert.notNull(lockRegistry, "The LockRegistry cannot be null"); this.idToMessage = new ConcurrentHashMap>(); this.groupIdToMessageGroup = new ConcurrentHashMap(); this.individualUpperBound = new UpperBound(individualCapacity); this.groupUpperBound = new UpperBound(groupCapacity); + this.lockRegistry = lockRegistry; } /** @@ -80,12 +97,19 @@ public class SimpleMessageStore extends AbstractMessageGroupStore implements Mes this(0); } + public void setLockRegistry(LockRegistry lockRegistry) { + Assert.notNull(lockRegistry, "The LockRegistry cannot be null"); + Assert.isTrue(!(this.isUsed), "Cannot change the lock registry after the store has been used"); + this.lockRegistry = lockRegistry; + } + @ManagedAttribute public long getMessageCount() { return idToMessage.size(); } public Message addMessage(Message message) { + this.isUsed = true; if (!individualUpperBound.tryAcquire(0)) { throw new MessagingException(this.getClass().getSimpleName() + " was out of capacity at, try constructing it with a larger capacity."); @@ -122,62 +146,112 @@ public class SimpleMessageStore extends AbstractMessageGroupStore implements Mes throw new MessagingException(this.getClass().getSimpleName() + " was out of capacity at, try constructing it with a larger capacity."); } - Object lock = this.obtainLock(groupId); - synchronized (lock) { - SimpleMessageGroup group = this.groupIdToMessageGroup.get(groupId); - if (group == null) { - group = new SimpleMessageGroup(groupId); - this.groupIdToMessageGroup.putIfAbsent(groupId, group); + Lock lock = this.lockRegistry.obtain(groupId); + try { + lock.lockInterruptibly(); + try { + SimpleMessageGroup group = this.groupIdToMessageGroup.get(groupId); + if (group == null) { + group = new SimpleMessageGroup(groupId); + this.groupIdToMessageGroup.putIfAbsent(groupId, group); + } + group.add(message); + return group; } - group.add(message); - return group; + finally { + lock.unlock(); + } + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new MessagingException("Interrupted while obtaining lock", e); } } public void removeMessageGroup(Object groupId) { - Object lock = this.obtainLock(groupId); - synchronized (lock) { - if (!groupIdToMessageGroup.containsKey(groupId)) { - return; + Lock lock = this.lockRegistry.obtain(groupId); + try { + lock.lockInterruptibly(); + try { + if (!groupIdToMessageGroup.containsKey(groupId)) { + return; + } + + groupUpperBound.release(groupIdToMessageGroup.get(groupId).size()); + groupIdToMessageGroup.remove(groupId); } - - groupUpperBound.release(groupIdToMessageGroup.get(groupId).size()); - groupIdToMessageGroup.remove(groupId); + finally { + lock.unlock(); + } + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new MessagingException("Interrupted while obtaining lock", e); } } public MessageGroup removeMessageFromGroup(Object groupId, Message messageToRemove) { - Object lock = this.obtainLock(groupId); - synchronized (lock) { - SimpleMessageGroup group = this.groupIdToMessageGroup.get(groupId); - Assert.notNull(group, "MessageGroup for groupId '" + groupId + "' " + - "can not be located while attempting to remove Message from the MessageGroup"); - group.remove(messageToRemove); - return group; + Lock lock = this.lockRegistry.obtain(groupId); + try { + lock.lockInterruptibly(); + try { + SimpleMessageGroup group = this.groupIdToMessageGroup.get(groupId); + Assert.notNull(group, "MessageGroup for groupId '" + groupId + "' " + + "can not be located while attempting to remove Message from the MessageGroup"); + group.remove(messageToRemove); + return group; + } + finally { + lock.unlock(); + } + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new MessagingException("Interrupted while obtaining lock", e); } } public Iterator iterator() { return new HashSet(groupIdToMessageGroup.values()).iterator(); } - + public void setLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber) { - Object lock = this.obtainLock(groupId); - synchronized (lock) { - SimpleMessageGroup group = this.groupIdToMessageGroup.get(groupId); - Assert.notNull(group, "MessageGroup for groupId '" + groupId + "' " + - "can not be located while attempting to set 'lastReleasedSequenceNumber'"); - group.setLastReleasedMessageSequenceNumber(sequenceNumber); + Lock lock = this.lockRegistry.obtain(groupId); + try { + lock.lockInterruptibly(); + try { + SimpleMessageGroup group = this.groupIdToMessageGroup.get(groupId); + Assert.notNull(group, "MessageGroup for groupId '" + groupId + "' " + + "can not be located while attempting to set 'lastReleasedSequenceNumber'"); + group.setLastReleasedMessageSequenceNumber(sequenceNumber); + } + finally { + lock.unlock(); + } + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new MessagingException("Interrupted while obtaining lock", e); } } public void completeGroup(Object groupId) { - Object lock = this.obtainLock(groupId); - synchronized (lock) { - SimpleMessageGroup group = this.groupIdToMessageGroup.get(groupId); - Assert.notNull(group, "MessageGroup for groupId '" + groupId + "' " + - "can not be located while attempting to complete the MessageGroup"); - group.complete(); + Lock lock = this.lockRegistry.obtain(groupId); + try { + lock.lockInterruptibly(); + try { + SimpleMessageGroup group = this.groupIdToMessageGroup.get(groupId); + Assert.notNull(group, "MessageGroup for groupId '" + groupId + "' " + + "can not be located while attempting to complete the MessageGroup"); + group.complete(); + } + finally { + lock.unlock(); + } + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new MessagingException("Interrupted while obtaining lock", e); } } @@ -189,18 +263,8 @@ public class SimpleMessageStore extends AbstractMessageGroupStore implements Mes if (message != null){ this.removeMessageFromGroup(groupId, message); } - } - return message; - } - - private Object obtainLock(Object groupId){ - Object lock = this.locks.get(groupId); - if (lock == null){ - Object newLock = new Object(); - Object previousLock = this.locks.putIfAbsent(groupId, newLock); - lock = (previousLock == null) ? newLock : previousLock; } - return lock; + return message; } public int messageGroupSize(Object groupId) {