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 d6dacb8e9a..e8b2abe747 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 @@ -21,6 +21,7 @@ import java.util.concurrent.locks.Lock; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; + import org.springframework.beans.factory.BeanFactory; import org.springframework.integration.Message; import org.springframework.integration.MessageChannel; @@ -126,6 +127,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH public void setMessageStore(MessageGroupStore store) { this.messageStore = store; store.registerMessageGroupExpiryCallback(new MessageGroupCallback() { + @Override public void execute(MessageGroupStore messageGroupStore, MessageGroup group) { forceComplete(group); } @@ -143,6 +145,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH sequenceAware = this.releaseStrategy instanceof SequenceSizeReleaseStrategy; } + @Override public void setOutputChannel(MessageChannel outputChannel) { Assert.notNull(outputChannel, "'outputChannel' must not be null"); this.outputChannel = outputChannel; @@ -267,17 +270,32 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH try { lock.lockInterruptibly(); try { + MessageGroup groupNow = group; /* - * Refetch the group because it might have changed while we were waiting on + * If the group argument is not already complete, + * re-fetch it because it might have changed while we were waiting on * its lock. If the last modified timestamp changed, defer the completion * because the selection condition may have changed such that the group - * would no longer be eligible. + * would no longer be eligible. If the timestamp changed, it's a completely new + * group and should not be reaped on this cycle. + * + * If the group argument is already complete, do not re-fetch. + * Note: not all message stores provide a direct reference to its internal + * group so the initial 'isComplete()` will only return true for those stores if + * the group was already complete at the time of its selection as a candidate. + * + * If the group is marked complete, only consider it + * for reaping if it's empty (and both timestamps are unaltered). */ - MessageGroup groupNow = this.messageStore.getMessageGroup( - group.getGroupId()); + if (!group.isComplete()) { + groupNow = this.messageStore.getMessageGroup(correlationKey); + } long lastModifiedNow = groupNow.getLastModified(); - if (group.getLastModified() == lastModifiedNow) { - if (groupNow.size() > 0) { + int groupSize = groupNow.size(); + if ((!groupNow.isComplete() || groupSize == 0) + && group.getLastModified() == lastModifiedNow + && group.getTimestamp() == groupNow.getTimestamp()) { + if (groupSize > 0) { if (releaseStrategy.canRelease(groupNow)) { this.completeGroup(correlationKey, groupNow); } @@ -305,7 +323,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH } } } - finally { + finally { if (removeGroup) { this.remove(group); } @@ -346,7 +364,8 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH + correlationKey + "] to: " + outputChannel); } completeGroup(correlationKey, group); - } else { + } + else { if (logger.isDebugEnabled()) { logger.debug("Discarding messages of partially complete group with key [" + correlationKey + "] to: " + discardChannel); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandlerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandlerTests.java index ecc647d3bd..c1f3a2d929 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandlerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandlerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2012 the original author or authors. + * Copyright 2002-2013 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. @@ -16,9 +16,13 @@ package org.springframework.integration.aggregator; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.verify; import java.lang.reflect.Method; import java.util.ArrayList; @@ -30,6 +34,7 @@ import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import org.junit.Test; + import org.springframework.beans.DirectFieldAccessor; import org.springframework.integration.Message; import org.springframework.integration.MessageChannel; @@ -57,6 +62,7 @@ public class AbstractCorrelatingMessageHandlerTests { AbstractCorrelatingMessageHandler handler = new AbstractCorrelatingMessageHandler( new MessageGroupProcessor() { + @Override public Object processMessageGroup(MessageGroup group) { return group; } @@ -73,6 +79,7 @@ public class AbstractCorrelatingMessageHandlerTests { */ Executors.newSingleThreadExecutor().execute(new Runnable() { + @Override public void run() { try { waitReapStartLatch.await(10, TimeUnit.SECONDS); @@ -98,6 +105,7 @@ public class AbstractCorrelatingMessageHandlerTests { /* * Executes when group 'bar' completes normally */ + @Override public boolean send(Message message, long timeout) { outputMessages.add(message); // wake reaper @@ -115,12 +123,14 @@ public class AbstractCorrelatingMessageHandlerTests { return true; } + @Override public boolean send(Message message) { return this.send(message, 0); } }); handler.setReleaseStrategy(new ReleaseStrategy() { + @Override public boolean canRelease(MessageGroup group) { return group.size() == 2; } @@ -162,6 +172,7 @@ public class AbstractCorrelatingMessageHandlerTests { AggregatingMessageHandler handler = new AggregatingMessageHandler( new MessageGroupProcessor() { + @Override public Object processMessageGroup(MessageGroup group) { return group; } @@ -174,17 +185,20 @@ public class AbstractCorrelatingMessageHandlerTests { /* * Executes when group 'bar' completes normally */ + @Override public boolean send(Message message, long timeout) { outputMessages.add(message); return true; } + @Override public boolean send(Message message) { return this.send(message, 0); } }); handler.setReleaseStrategy(new ReleaseStrategy() { + @Override public boolean canRelease(MessageGroup group) { return group.size() == 1; } @@ -208,6 +222,7 @@ public class AbstractCorrelatingMessageHandlerTests { AggregatingMessageHandler handler = new AggregatingMessageHandler( new MessageGroupProcessor() { + @Override public Object processMessageGroup(MessageGroup group) { return group; } @@ -220,17 +235,20 @@ public class AbstractCorrelatingMessageHandlerTests { /* * Executes when group 'bar' completes normally */ + @Override public boolean send(Message message, long timeout) { outputMessages.add(message); return true; } + @Override public boolean send(Message message) { return this.send(message, 0); } }); handler.setReleaseStrategy(new ReleaseStrategy() { + @Override public boolean canRelease(MessageGroup group) { return group.size() == 1; } @@ -258,6 +276,7 @@ public class AbstractCorrelatingMessageHandlerTests { MessageGroupProcessor mgp = new DefaultAggregatingMessageGroupProcessor(); AggregatingMessageHandler handler = new AggregatingMessageHandler(mgp); handler.setReleaseStrategy(new ReleaseStrategy() { + @Override public boolean canRelease(MessageGroup group) { return true; } @@ -283,4 +302,101 @@ public class AbstractCorrelatingMessageHandlerTests { assertEquals(1, payload.size()); } + @Test /* INT-3216 */ + public void testDontReapIfAlreadyComplete() throws Exception { + MessageGroupProcessor mgp = new DefaultAggregatingMessageGroupProcessor(); + AggregatingMessageHandler handler = new AggregatingMessageHandler(mgp); + handler.setReleaseStrategy(new ReleaseStrategy() { + + @Override + public boolean canRelease(MessageGroup group) { + return true; + } + + }); + QueueChannel outputChannel = new QueueChannel(); + handler.setOutputChannel(outputChannel); + MessageGroupStore mgs = TestUtils.getPropertyValue(handler, "messageStore", MessageGroupStore.class); + mgs.addMessageToGroup("foo", new GenericMessage("foo")); + mgs.completeGroup("foo"); + mgs = spy(mgs); + new DirectFieldAccessor(handler).setPropertyValue("messageStore", mgs); + Method forceComplete = AbstractCorrelatingMessageHandler.class.getDeclaredMethod("forceComplete", MessageGroup.class); + forceComplete.setAccessible(true); + MessageGroup group = (MessageGroup) TestUtils.getPropertyValue(mgs, "groupIdToMessageGroup", Map.class).get("foo"); + assertTrue(group.isComplete()); + forceComplete.invoke(handler, group); + verify(mgs, never()).getMessageGroup("foo"); + assertNull(outputChannel.receive(0)); + } + + /* + * INT-3216 - Verifies the complete early exit is taken after a refresh. + */ + @Test + public void testDontReapIfAlreadyCompleteAfterRefetch() throws Exception { + MessageGroupProcessor mgp = new DefaultAggregatingMessageGroupProcessor(); + AggregatingMessageHandler handler = new AggregatingMessageHandler(mgp); + handler.setReleaseStrategy(new ReleaseStrategy() { + + @Override + public boolean canRelease(MessageGroup group) { + return true; + } + + }); + QueueChannel outputChannel = new QueueChannel(); + handler.setOutputChannel(outputChannel); + MessageGroupStore mgs = TestUtils.getPropertyValue(handler, "messageStore", MessageGroupStore.class); + mgs.addMessageToGroup("foo", new GenericMessage("foo")); + MessageGroup group = mgs.getMessageGroup("foo"); + mgs.completeGroup("foo"); + mgs = spy(mgs); + new DirectFieldAccessor(handler).setPropertyValue("messageStore", mgs); + Method forceComplete = AbstractCorrelatingMessageHandler.class.getDeclaredMethod("forceComplete", MessageGroup.class); + forceComplete.setAccessible(true); + MessageGroup groupInStore = (MessageGroup) TestUtils.getPropertyValue(mgs, "groupIdToMessageGroup", Map.class).get("foo"); + assertTrue(groupInStore.isComplete()); + assertFalse(group.isComplete()); + new DirectFieldAccessor(group).setPropertyValue("lastModified", groupInStore.getLastModified()); + forceComplete.invoke(handler, group); + verify(mgs).getMessageGroup("foo"); + assertNull(outputChannel.receive(0)); + } + + /* + * INT-3216 - Verifies we don't complete if it's a completely new group (different timestamp). + */ + @Test + public void testDontReapIfNewGroupFoundDuringRefetch() throws Exception { + MessageGroupProcessor mgp = new DefaultAggregatingMessageGroupProcessor(); + AggregatingMessageHandler handler = new AggregatingMessageHandler(mgp); + handler.setReleaseStrategy(new ReleaseStrategy() { + + @Override + public boolean canRelease(MessageGroup group) { + return true; + } + + }); + QueueChannel outputChannel = new QueueChannel(); + handler.setOutputChannel(outputChannel); + MessageGroupStore mgs = TestUtils.getPropertyValue(handler, "messageStore", MessageGroupStore.class); + mgs.addMessageToGroup("foo", new GenericMessage("foo")); + MessageGroup group = mgs.getMessageGroup("foo"); + mgs = spy(mgs); + new DirectFieldAccessor(handler).setPropertyValue("messageStore", mgs); + Method forceComplete = AbstractCorrelatingMessageHandler.class.getDeclaredMethod("forceComplete", MessageGroup.class); + forceComplete.setAccessible(true); + MessageGroup groupInStore = (MessageGroup) TestUtils.getPropertyValue(mgs, "groupIdToMessageGroup", Map.class).get("foo"); + assertFalse(groupInStore.isComplete()); + assertFalse(group.isComplete()); + DirectFieldAccessor directFieldAccessor = new DirectFieldAccessor(group); + directFieldAccessor.setPropertyValue("lastModified", groupInStore.getLastModified()); + directFieldAccessor.setPropertyValue("timestamp", groupInStore.getTimestamp() - 1); + forceComplete.invoke(handler, group); + verify(mgs).getMessageGroup("foo"); + assertNull(outputChannel.receive(0)); + } + }