diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java index 0926f40b32..443d50853a 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java @@ -34,9 +34,16 @@ import java.util.concurrent.*; import java.util.concurrent.locks.ReentrantLock; /** - * MessageHandler that holds a buffer of messages in a MessageStore. This class takes care of - * correlated groups of messages that can be completed in batches. It is useful for aggregating, - * resequencing, or custom implementations requiring correlation. + * MessageHandler that holds a buffer of correlated messages in a MessageStore. This class takes care of correlated + * groups of messages that can be completed in batches. It is useful for aggregating, resequencing, or custom + * implementations requiring correlation. + *

+ * To customize this handler inject {@link org.springframework.integration.aggregator.CorrelationStrategy}, {@link + * org.springframework.integration.aggregator.CompletionStrategy} and {@link org.springframework.integration.aggregator.MessageGroupProcessor} + * implementations as you require. + *

+ * By default the CorrelationStrategy will be a HeaderAttributeCorrelationStrategy and the CompletionStrategy will be a + * SequenceSizeCompletionStrategy. * * @author Iwein Fuld */ @@ -115,9 +122,10 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements protected void handleMessageInternal(Message message) throws Exception { Object correlationKey = correlationStrategy.getCorrelationKey(message); try { - if (tracker.aquireLockFor(correlationKey)) { + if (tracker.waitForLockIfNotTracked(correlationKey)) { MessageGroup group = - new MessageGroup(store, completionStrategy, correlationKey, deleteOrTrackCallback()); + new MessageGroup(store.list(correlationKey), + completionStrategy, correlationKey, deleteOrTrackCallback()); if (group.hasNoMessageSuperseding(message)) { store(message, correlationKey); group.add(message); @@ -146,7 +154,7 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements } public void onCompletionOf(Object correlationKey) { - tracker.pushCorrellationId(correlationKey); + tracker.pushCorrelationId(correlationKey); } }; } @@ -195,7 +203,9 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements if (logger.isDebugEnabled()) { logger.debug(this + "'s PrunerTask is processing " + key); } - forceComplete(key); + if (!forceComplete(key)) { + keysInBuffer.offer(delayedKey); + } } } catch (InterruptedException e) { Thread.currentThread().interrupt(); @@ -204,34 +214,42 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements } } - protected final void forceComplete(Object key) { - MessageGroup group = new MessageGroup(store, completionStrategy, key, deleteOrTrackCallback()); - List> all = store.list(key); - if (all.size() > 0) { - //last chance for normal completion - MessageChannel outputChannel = resolveReplyChannel(all.get(0), this.outputChannel, this.channelResolver); - boolean processed = false; - if (completionStrategy.isComplete(all)) { - outputProcessor.processAndSend(group, outputChannel); - processed = true; - } - boolean fullyCompleted = processed; - if (!fullyCompleted) { - if (sendPartialResultOnTimeout) { - if (logger.isInfoEnabled()) { - logger.info("Processing partially complete messages for key [" + key + "] to: " + outputChannel); + protected final boolean forceComplete(Object key) { + try { + if (tracker.tryLockFor(key)) { + List> all = store.list(key); + MessageGroup group = new MessageGroup(all, completionStrategy, key, deleteOrTrackCallback()); + if (all.size() > 0) { + //last chance for normal completion + MessageChannel outputChannel = resolveReplyChannel(all.get(0), this.outputChannel, this.channelResolver); + boolean processed = false; + if (group.isComplete()) { + outputProcessor.processAndSend(group, outputChannel); + processed = true; } - outputProcessor.processAndSend(group, outputChannel); - } else { - if (logger.isInfoEnabled()) { - logger.info("Discarding partially complete messages for key [" + key + "] to: " + discardChannel); - } - for (Message message : all) { - discardChannel.send(message); - store.delete(message.getHeaders().getId()); + if (!processed) { + if (sendPartialResultOnTimeout) { + if (logger.isInfoEnabled()) { + logger.info("Processing partially complete messages for key [" + key + "] to: " + outputChannel); + } + outputProcessor.processAndSend(group, outputChannel); + } else { + if (logger.isInfoEnabled()) { + logger.info("Discarding partially complete messages for key [" + key + "] to: " + discardChannel); + } + for (Message message : all) { + discardChannel.send(message); + store.delete(message.getHeaders().getId()); + } + } } } + return true; + } else { + return false; } + } finally { + tracker.unlock(key); } } @@ -262,20 +280,27 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements private final class IdTracker { private ConcurrentMap trackerLocks = new ConcurrentHashMap(); - private final Queue trackedCorrellationIds = new LinkedBlockingQueue(); + private final Queue trackedCorrelationIds = new LinkedBlockingQueue(); - private void pushCorrellationId(Object correlationKey) { - while (!trackedCorrellationIds.offer(correlationKey)) { + private void pushCorrelationId(Object correlationKey) { + while (!trackedCorrelationIds.offer(correlationKey)) { //make room in the queue - trackedCorrellationIds.poll(); + trackedCorrelationIds.poll(); } trackerLocks.remove(correlationKey); } - private boolean aquireLockFor(Object correlationKey) { + /** + * Call this method to check if an id is tracked and obtain a lock for it. Don't forget to finally unlock + * afterwards. + * + * @return false if the key was tracked, true after obtaining the lock otherwise. + */ + private boolean waitForLockIfNotTracked(Object correlationKey) { ReentrantLock lock = trackerLocks.get(correlationKey); if (lock == null) { - if (trackedCorrellationIds.contains(correlationKey)) { + if (trackedCorrelationIds.contains(correlationKey)) { + //this correlation key is already processed in the near past: disallow processing return false; } lock = new ReentrantLock(); @@ -286,6 +311,16 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements return true; } + private boolean tryLockFor(Object correlationKey) { + ReentrantLock lock = trackerLocks.get(correlationKey); + if (lock == null) { + lock = new ReentrantLock(); + ReentrantLock original = trackerLocks.putIfAbsent(correlationKey, lock); + lock = original == null ? lock : original; + } + return lock.tryLock(); + } + private void unlock(Object correlationKey) { ReentrantLock lock = trackerLocks.get(correlationKey); if (lock != null && lock.isHeldByCurrentThread()) { diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageGroup.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageGroup.java index 11d21d96dc..b3607fe02a 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageGroup.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageGroup.java @@ -1,38 +1,36 @@ package org.springframework.integration.aggregator; import org.springframework.integration.core.Message; -import org.springframework.integration.store.MessageStore; -import java.util.ArrayList; -import java.util.Arrays; -import java.util.Collections; -import java.util.List; +import java.util.*; /** - * Represents a group of correlated messages that is bound to a certain - * {@link org.springframework.integration.store.MessageStore} and correlation key. - * The group can grow during its lifetime, if messages are added to it. + * Represents an mutable group of correlated messages that is bound to a certain {@link org.springframework.integration.store.MessageStore} + * and correlation key. The group will grow during its lifetime, when messages are added to it. + * This is not thread safe and should not be used for long running aggregations. *

- * According to its {@link org.springframework.integration.aggregator.CompletionStrategy} it - * can be complete depending on the messages in the group. + * According to its {@link org.springframework.integration.aggregator.CompletionStrategy} it can be complete + * depending on the messages in the group. *

- * Listeners can be configured to get callbacks when (parts of) the group are processed or the whole group is completed. + * Optionally MessageGroupListeners can be added to get callbacks when (parts of) the group are processed or the whole + * group is completed. + * + * * * @author Iwein Fuld */ public class MessageGroup { - private final MessageStore store; private final CompletionStrategy completionStrategy; private final Object correlationKey; private final ArrayList> messages = new ArrayList>(); private final List listeners; - public MessageGroup(MessageStore store, CompletionStrategy completionStrategy, Object correlationKey, MessageGroupListener... listeners) { - this.store = store; + public MessageGroup(Collection> originalMessages, CompletionStrategy completionStrategy, Object correlationKey, + MessageGroupListener... listeners) { this.completionStrategy = completionStrategy; this.correlationKey = correlationKey; - this.messages.addAll(store.list(correlationKey)); + this.messages.addAll(originalMessages); this.listeners = Collections.unmodifiableList(Arrays.asList(listeners)); } @@ -53,10 +51,6 @@ public class MessageGroup { return completionStrategy.isComplete(messages); } - public static MessageGroupBuilder builder() { - return new MessageGroupBuilder(); - } - public List> getMessages() { return messages; } @@ -76,35 +70,4 @@ public class MessageGroup { listener.onCompletionOf(correlationKey); } } - - final static class MessageGroupBuilder { - private MessageStore store; - private Object correlationKey; - private MessageGroupListener[] listeners; - private CompletionStrategy completionStrategy; - - public MessageGroupBuilder withCorrelationKey(Object key) { - this.correlationKey = key; - return this; - } - - public MessageGroupBuilder withStore(MessageStore store) { - this.store = store; - return this; - } - - public MessageGroupBuilder observedBy(MessageGroupListener... listeners) { - this.listeners = listeners; - return this; - } - - public MessageGroupBuilder completedBy(CompletionStrategy completionStrategy) { - this.completionStrategy = completionStrategy; - return this; - } - - public MessageGroup build() { - return new MessageGroup(store, completionStrategy, correlationKey, listeners); - } - } } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerIntegrationTest.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerIntegrationTest.java index cbd3ea9cd2..43ff47dabe 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerIntegrationTest.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerIntegrationTest.java @@ -1,3 +1,19 @@ +/* + * Copyright 2002-2009 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. + */ + package org.springframework.integration.aggregator; import org.junit.Before; diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java index 80f97cb493..27ac2d4de4 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java @@ -20,7 +20,9 @@ import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; import org.mockito.Mock; +import org.mockito.invocation.InvocationOnMock; import org.mockito.runners.MockitoJUnitRunner; +import org.mockito.stubbing.Answer; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.core.MessageHeaders; @@ -30,7 +32,10 @@ import org.springframework.integration.store.MessageStore; import java.util.ArrayList; import java.util.Arrays; import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; +import static org.junit.Assert.assertFalse; import static org.mockito.Matchers.eq; import static org.mockito.Matchers.isA; import static org.mockito.Mockito.*; @@ -41,68 +46,123 @@ import static org.mockito.Mockito.*; @RunWith(MockitoJUnitRunner.class) public class CorrelatingMessageHandlerTests { - private CorrelatingMessageHandler handler; + private CorrelatingMessageHandler handler; - @Mock - private MessageStore store; + @Mock + private MessageStore store; - @Mock - private CorrelationStrategy correlationStrategy; + @Mock + private CorrelationStrategy correlationStrategy; - @Mock - private CompletionStrategy completionStrategy; + @Mock + private CompletionStrategy completionStrategy; - @Mock - private MessageGroupProcessor processor; + @Mock + private MessageGroupProcessor processor; - @Mock - private MessageChannel outputChannel; + @Mock + private MessageChannel outputChannel; - @Before - public void initializeSubject() { - handler = new CorrelatingMessageHandler( - store, correlationStrategy, completionStrategy, processor); - handler.setOutputChannel(outputChannel); - } + @Before + public void initializeSubject() { + handler = new CorrelatingMessageHandler( + store, correlationStrategy, completionStrategy, processor); + handler.setOutputChannel(outputChannel); + doAnswer(new Answer() { + public Object answer(InvocationOnMock invocation) throws Throwable { + MessageGroup messageGroup = (MessageGroup) invocation.getArguments()[0]; + messageGroup.onProcessingOf(messageGroup.getMessages().toArray(new Message[2])); + messageGroup.onCompletion(); + return null; + } + }).when(processor).processAndSend(isA(MessageGroup.class), + eq(outputChannel)); + } - @Test - public void bufferCompletesNormally() throws Exception { - String correlationKey = "key"; - Message message1 = testMessage(1, 1); - Message message2 = testMessage(2, 2); - List> storedMessages = new ArrayList>(); + @Test + public void bufferCompletesNormally() throws Exception { + String correlationKey = "key"; + Message message1 = testMessage(1, 1); + Message message2 = testMessage(2, 2); + List> storedMessages = new ArrayList>(); - when(store.list(correlationKey)).thenReturn(storedMessages); + when(store.list(correlationKey)).thenReturn(storedMessages); - when(correlationStrategy.getCorrelationKey(isA(Message.class))) - .thenReturn(correlationKey); + when(correlationStrategy.getCorrelationKey(isA(Message.class))) + .thenReturn(correlationKey); - when(completionStrategy.isComplete(Arrays.asList(message1))).thenReturn(false); + when(completionStrategy.isComplete(Arrays.asList(message1))).thenReturn(false); - handler.handleMessage(message1); - storedMessages.add(message1); + handler.handleMessage(message1); + storedMessages.add(message1); - when(completionStrategy.isComplete(Arrays.asList(message1, message2))).thenReturn(true); - handler.handleMessage(message2); - storedMessages.add(message2); + when(completionStrategy.isComplete(Arrays.asList(message1, message2))).thenReturn(true); + handler.handleMessage(message2); + storedMessages.add(message2); - verify(store).put(message1); - verify(store).put(message2); - verify(store, times(2)).list(correlationKey); - verify(correlationStrategy).getCorrelationKey(message1); - verify(correlationStrategy).getCorrelationKey(message2); - verify(completionStrategy).isComplete(Arrays.asList(message1)); - verify(completionStrategy).isComplete(Arrays.asList(message1, message2)); - verify(processor).processAndSend(isA(MessageGroup.class), + verify(store).put(message1); + verify(store).put(message2); + verify(store, times(2)).list(correlationKey); + verify(correlationStrategy).getCorrelationKey(message1); + verify(correlationStrategy).getCorrelationKey(message2); + verify(completionStrategy).isComplete(Arrays.asList(message1)); + verify(completionStrategy).isComplete(Arrays.asList(message1, message2)); + verify(processor).processAndSend(isA(MessageGroup.class), eq(outputChannel) ); - } + } + /* + The next test verifies that when pruning happens after the completing message arrived, but before the group was + processed locking prevents forced completion and the group completes normally. + */ - private Message testMessage(int id, int sequenceNumber) { - return MessageBuilder.withPayload("test"+id) - .setHeader(MessageHeaders.ID, id) - .setSequenceNumber(sequenceNumber).build(); - } + @Test + public void shouldNotPruneWhileCompleting() throws Exception { + String correlationKey = "key"; + final Message message1 = testMessage(1, 1); + final Message message2 = testMessage(2, 2); + final List> storedMessages = new ArrayList>(); + + final CountDownLatch bothMessagesHandled = new CountDownLatch(2); + + when(store.list(correlationKey)).thenReturn(storedMessages); + + when(correlationStrategy.getCorrelationKey(isA(Message.class))) + .thenReturn(correlationKey); + + when(completionStrategy.isComplete(Arrays.asList(message1, message2))).thenAnswer(new Answer() { + public Boolean answer(InvocationOnMock invocation) throws Throwable { + Thread.sleep(50); + return true; + } + }).thenReturn(true); + + handler.handleMessage(message1); + bothMessagesHandled.countDown(); + storedMessages.add(message1); + Executors.newSingleThreadExecutor().submit(new Runnable() { + public void run() { + handler.handleMessage(message2); + storedMessages.add(message2); + bothMessagesHandled.countDown(); + } + }); + + Thread.sleep(10); + assertFalse(handler.forceComplete("key")); + + bothMessagesHandled.await(); + verify(store).put(message1); + verify(store).put(message2); + verify(store).delete(1); + verify(store).delete(2); + } + + private Message testMessage(int id, int sequenceNumber) { + return MessageBuilder.withPayload("test" + id) + .setHeader(MessageHeaders.ID, id) + .setSequenceNumber(sequenceNumber).build(); + } } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/MessageGroupTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/MessageGroupTests.java index 8b1f10f259..1458c56e5c 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/MessageGroupTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/MessageGroupTests.java @@ -1,11 +1,19 @@ package org.springframework.integration.aggregator; +import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; import org.mockito.Mock; import org.mockito.runners.MockitoJUnitRunner; +import org.springframework.integration.core.Message; +import org.springframework.integration.message.MessageBuilder; import org.springframework.integration.store.MessageStore; +import java.util.Collections; + +import static org.hamcrest.CoreMatchers.is; +import static org.junit.Assert.assertThat; + /** * */ @@ -23,11 +31,19 @@ public class MessageGroupTests { @Mock private CompletionStrategy completionStrategy; + private MessageGroup group; + + @Before + public void buildMessageGroup() { + group = new MessageGroup(Collections.>emptyList(), completionStrategy, key, listener); + } + @Test - public void shouldBuildMessageGroup() { - MessageGroup group = MessageGroup.builder(). - withStore(store).withCorrelationKey(key). - completedBy(completionStrategy).observedBy(listener). - build(); + public void shouldFindSupersedingMessages() { + final Message message1 = MessageBuilder.withPayload("test").setCorrelationId("foo").build(); + final Message message2 = MessageBuilder.fromMessage(message1).build(); + assertThat(group.hasNoMessageSuperseding(message1), is(true)); + group.add(message2); + assertThat(group.hasNoMessageSuperseding(message1), is(false)); } } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/NewConcurrentAggregatorEndpointTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/NewConcurrentAggregatorEndpointTests.java index 8051368f6e..ae7958d931 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/NewConcurrentAggregatorEndpointTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/NewConcurrentAggregatorEndpointTests.java @@ -73,7 +73,8 @@ public class NewConcurrentAggregatorEndpointTests { this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, message1, latch)); this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, message2, latch)); this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, message3, latch)); - latch.await(1000, TimeUnit.MILLISECONDS); + latch.await(10000, TimeUnit.MILLISECONDS); + assertThat(latch.getCount(), is(0l)); Message reply = replyChannel.receive(2000); assertNotNull(reply); assertEquals(reply.getPayload(), 105); @@ -184,7 +185,7 @@ public class NewConcurrentAggregatorEndpointTests { assertEquals(3, replyChannel.receive(100).getPayload()); this.aggregator.handleMessage(createMessage(4, 3, 1, 1, replyChannel, null)); assertEquals(4, replyChannel.receive(100).getPayload()); - //next message with same correllation ID is discarded + //next message with same correlation ID is discarded this.aggregator.handleMessage(createMessage(2, 1, 1, 1, replyChannel, null)); assertEquals(2, discardChannel.receive(100).getPayload()); } @@ -232,7 +233,7 @@ public class NewConcurrentAggregatorEndpointTests { this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, message3, latch)); this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, message4, latch)); latch.await(1000, TimeUnit.MILLISECONDS); - Message reply = replyChannel.receive(0); + Message reply = replyChannel.receive(100); assertNotNull("A message should be aggregated", reply); assertThat(((Integer) reply.getPayload()), is(105)); } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/scheduling/CronTriggerTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/scheduling/CronTriggerTests.java index 156ba15fd4..c973b95be5 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/scheduling/CronTriggerTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/scheduling/CronTriggerTests.java @@ -16,17 +16,17 @@ package org.springframework.integration.scheduling; -import static org.junit.Assert.assertEquals; +import org.junit.Before; +import org.junit.Ignore; +import org.junit.Test; +import org.springframework.scheduling.support.CronTrigger; +import org.springframework.scheduling.support.SimpleTriggerContext; import java.util.Calendar; import java.util.Date; import java.util.GregorianCalendar; -import org.junit.Before; -import org.junit.Test; - -import org.springframework.scheduling.support.CronTrigger; -import org.springframework.scheduling.support.SimpleTriggerContext; +import static org.junit.Assert.assertEquals; /** * @author Dave Syer @@ -239,6 +239,7 @@ public class CronTriggerTests { } @Test + @Ignore //see INT-971 public void testDailyTriggerInLongMonth() throws Exception { CronTrigger trigger = new CronTrigger("0 0 0 * * *"); calendar.set(Calendar.MONTH, 9); // October: 31 days