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

Merge pull request #463 from garyrussell/INT-2592

Browse Source 
* INT-2592:
  INT-2592 Fix Memory Leak in SimpleMessageStore
This commit is contained in:
Oleg Zhurakousky
2012-06-01 13:15:22 -04:00
parent 4bc8760012 e053932c5c
commit 0646df3657
2 changed files with 144 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files

41
spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java
View File

@@ -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.
* <p/>
* 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<Message<?>> 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<Message<?>> partialSequence){
List<Message<?>> sorted = new ArrayList<Message<?>>((Collection<? extends Message<?>>)partialSequence);
List<Message<?>> sorted = new ArrayList<Message<?>>(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<Message<?>> 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;

176
spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java
View File

@@ -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<Object, Object> locks = new ConcurrentHashMap<Object, Object>();
private volatile LockRegistry lockRegistry;
private final ConcurrentMap<UUID, Message<?>> idToMessage;
private final ConcurrentMap<Object, SimpleMessageGroup> 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<UUID, Message<?>>();
this.groupIdToMessageGroup = new ConcurrentHashMap<Object, SimpleMessageGroup>();
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 <T> Message<T> addMessage(Message<T> 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<MessageGroup> iterator() {
return new HashSet<MessageGroup>(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) {