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

INT-330:

Browse Source 
- Polished some javadoc
 - removed unnecessary dependency on MessageStore from MessageGroup
 - added locking in pruner

INT-971: disabled test from CronTriggerTests to avoid build breakage pending fix.
This commit is contained in:
Iwein Fuld
2010-01-28 10:24:36 +00:00
parent 095f31d61c
commit a16e4aa6bd
7 changed files with 239 additions and 147 deletions
Download Patch File Download Diff File Expand all files Collapse all files

109
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java
View File

@@ -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.
* <p/>
* 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.
* <p/>
* 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<Message<?>> 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<Message<?>> 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<Object, ReentrantLock> trackerLocks = new ConcurrentHashMap<Object, ReentrantLock>();
private final Queue<Object> trackedCorrellationIds = new LinkedBlockingQueue<Object>();
private final Queue<Object> trackedCorrelationIds = new LinkedBlockingQueue<Object>();
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()) {

63
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageGroup.java
View File

@@ -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 <code>add</code>ed 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 <code>add</code>ed to it. <strong>
* This is not thread safe and should not be used for long running aggregations</strong>.
* <p/>
* According to its {@link org.springframework.integration.aggregator.CompletionStrategy} it
* can be <i>complete</i> depending on the messages in the group.
* According to its {@link org.springframework.integration.aggregator.CompletionStrategy} it can be <i>complete</i>
* depending on the messages in the group.
* <p/>
* 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<Message<?>> messages = new ArrayList<Message<?>>();
private final List<MessageGroupListener> listeners;
public MessageGroup(MessageStore store, CompletionStrategy completionStrategy, Object correlationKey, MessageGroupListener... listeners) {
this.store = store;
public MessageGroup(Collection<? extends Message<?>> 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<Message<?>> 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);
}
}
}

16
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerIntegrationTest.java
View File

@@ -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;

152
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java
View File

@@ -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<Message<?>> storedMessages = new ArrayList<Message<?>>();
@Test
public void bufferCompletesNormally() throws Exception {
String correlationKey = "key";
Message<?> message1 = testMessage(1, 1);
Message<?> message2 = testMessage(2, 2);
List<Message<?>> storedMessages = new ArrayList<Message<?>>();
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<Message<?>> storedMessages = new ArrayList<Message<?>>();
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<Boolean>() {
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();
}
}

26
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/MessageGroupTests.java
View File

@@ -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.<Message<?>>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));
}
}

7
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/NewConcurrentAggregatorEndpointTests.java
View File

@@ -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));
}

13
org.springframework.integration/src/test/java/org/springframework/integration/scheduling/CronTriggerTests.java
View File

@@ -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