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

INT-330: moved base components for new aggregation over to HEAD, also fixed .classpaths

Browse Source
This commit is contained in:
Iwein Fuld
2009-09-21 13:56:05 +00:00
parent 2bfa7cb42a
commit 942579e8a7
26 changed files with 1652 additions and 152 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/BufferedMessagesCallback.java Normal file
View File

@@ -0,0 +1,28 @@
/*
* 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.springframework.integration.core.Message;
/**
* @author Iwein Fuld
*/
public interface BufferedMessagesCallback {
void onProcessingOf(Message<?>... processedMessages);
void onCompletionOf(Object correlationKey);
}

276
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/BufferingMessageHandler.java Normal file
View File

@@ -0,0 +1,276 @@
/*
* 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.springframework.context.Lifecycle;
import org.springframework.integration.aggregator.*;
import org.springframework.integration.channel.ChannelResolutionException;
import org.springframework.integration.channel.ChannelResolver;
import org.springframework.integration.channel.NullChannel;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.core.MessageHeaders;
import org.springframework.integration.handler.AbstractMessageHandler;
import org.springframework.integration.store.MessageStore;
import org.springframework.integration.store.SimpleMessageStore;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.util.Assert;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.*;
/**
* MessageHandler that holds a buffer of messages in a MessageStore
*
* @author Iwein Fuld
*/
public class BufferingMessageHandler extends AbstractMessageHandler implements Lifecycle {
private MessageStore store = new SimpleMessageStore(100);
private final CorrelationStrategy correlationStrategy;
//TODO decide if we still support tracking capacity, and if this needs to be moved into the Store instead
private final Queue trackedCorrellationIds = new LinkedBlockingQueue();
private final CompletionStrategy completionStrategy;
private MessagesProcessor outputProcessor;
private MessageChannel outputChannel;
private volatile MessageChannel discardChannel = new NullChannel();
private TaskScheduler taskScheduler;
private Object lifecycleMonitor = new Object();
private ScheduledFuture reaperFutureTask;
private volatile long reaperInterval = 1000l;
private final BlockingQueue<DelayedKey> keysInBuffer = new DelayQueue<DelayedKey>();
private volatile long timeout = 60000l;
private volatile boolean sendPartialResultOnTimeout;
private ChannelResolver channelResolver;
public BufferingMessageHandler(MessageStore store,
CorrelationStrategy correlationStrategy,
CompletionStrategy completionStrategy, MessagesProcessor processor
) {
Assert.notNull(store);
Assert.notNull(correlationStrategy);
Assert.notNull(completionStrategy);
Assert.notNull(processor);
this.store = store;
this.correlationStrategy = correlationStrategy;
this.completionStrategy = completionStrategy;
this.outputProcessor = processor;
}
public BufferingMessageHandler(MessageStore store,
MessagesProcessor processor) {
this(store, new HeaderAttributeCorrelationStrategy(
MessageHeaders.CORRELATION_ID),
new SequenceSizeCompletionStrategy(), processor);
}
public void setTaskScheduler(TaskScheduler taskScheduler) {
this.taskScheduler = taskScheduler;
}
public void setTimeout(long timeout) {
this.timeout = timeout;
}
public void setReaperInterval(long reaperInterval) {
this.reaperInterval = reaperInterval;
}
public void setOutputChannel(MessageChannel outputChannel) {
this.outputChannel = outputChannel;
}
public void setChannelResolver(ChannelResolver channelResolver) {
this.channelResolver = channelResolver;
}
public void setDiscardChannel(MessageChannel discardChannel) {
this.discardChannel = discardChannel;
}
public void setSendPartialResultOnTimeout(boolean sendPartialResultOnTimeout) {
this.sendPartialResultOnTimeout = sendPartialResultOnTimeout;
}
@Override
protected void handleMessageInternal(Message<?> message) throws Exception {
Object correlationKey = correlationStrategy.getCorrelationKey(message);
if (!trackedCorrellationIds.contains(correlationKey)) {
store(message, correlationKey);
List<Message<?>> all = store.getAll(correlationKey);
complete(correlationKey, all, this.resolveReplyChannel(message));
} else {
discardChannel.send(message);
}
}
private boolean complete(Object correlationKey, List<Message<?>> correlatedMessages, MessageChannel messageChannel) {
boolean processed = false;
if (completionStrategy.isComplete(correlatedMessages)) {
outputProcessor.processAndSend(correlationKey, correlatedMessages, messageChannel, deleteOrTrackCallback());
processed = true;
}
return processed;
}
private void pushCorrellationId(Queue trackedCorrellationIds, Object correlationKey) {
while (!trackedCorrellationIds.offer(correlationKey)) {
//make room in the queue
trackedCorrellationIds.poll();
}
}
private BufferedMessagesCallback deleteOrTrackCallback() {
return new BufferedMessagesCallback() {
public void onProcessingOf(Message<?>... processedMessage) {
for (Message<?> message : processedMessage) {
store.delete(message.getHeaders().getId());
}
}
public void onCompletionOf(Object correlationKey) {
pushCorrellationId(trackedCorrellationIds, correlationKey);
}
};
}
private void store(Message<?> message, Object correlationKey) {
store.put(message);
if (!keysInBuffer.contains(correlationKey)) {
keysInBuffer.add(new DelayedKey(correlationKey, timeout));
}
}
//TODO copied from AbstractReplyProducingMessageHandler
private MessageChannel resolveReplyChannel(Message<?> requestMessage) {
MessageChannel replyChannel = outputChannel;
if (replyChannel == null) {
Object replyChannelHeader = requestMessage.getHeaders().getReplyChannel();
if (replyChannelHeader != null) {
if (replyChannelHeader instanceof MessageChannel) {
replyChannel = (MessageChannel) replyChannelHeader;
} else if (replyChannelHeader instanceof String) {
Assert.state(this.channelResolver != null,
"ChannelResolver is required for resolving a reply channel by name");
replyChannel = this.channelResolver.resolveChannelName((String) replyChannelHeader);
} else {
throw new ChannelResolutionException("expected a MessageChannel or String for 'replyChannel', but type is ["
+ replyChannelHeader.getClass() + "]");
}
}
}
if (replyChannel == null) {
throw new ChannelResolutionException(
"unable to resolve reply channel for message: " + requestMessage);
}
return replyChannel;
}
public boolean isRunning() {
synchronized (this.lifecycleMonitor) {
return this.reaperFutureTask != null;
}
}
public void start() {
synchronized (this.lifecycleMonitor) {
if (this.isRunning()) {
return;
}
Assert.state(this.taskScheduler != null, "'taskScheduler' must not be null");
this.reaperFutureTask = this.taskScheduler.scheduleWithFixedDelay(
new PrunerTask(), this.reaperInterval);
}
}
public void stop() {
synchronized (this.lifecycleMonitor) {
if (this.isRunning()) {
this.reaperFutureTask.cancel(true);
}
}
}
private class PrunerTask implements Runnable {
public void run() {
if (logger.isTraceEnabled()) {
logger.trace("PrunerTask is running");
}
DelayedKey delayedKey;
try {
while ((delayedKey = keysInBuffer.poll(reaperInterval, TimeUnit.MILLISECONDS)) != null) {
Object key = delayedKey.getKey();
if (logger.isDebugEnabled()) {
logger.debug(this + "'s PrunerTask is processing " + key);
}
forceComplete(key);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}
protected final void forceComplete(Object key) {
List<Message<?>> all = store.getAll(key);
if (all.size() > 0) {
//last chance for normal completion
MessageChannel outputChannel = resolveReplyChannel(all.get(0));
boolean fullyCompleted = complete(key, all, outputChannel);
if (!fullyCompleted) {
if (sendPartialResultOnTimeout) {
if (logger.isInfoEnabled()) {
logger.info("Processing partially complete messages for key [" + key + "] to: " + outputChannel);
}
outputProcessor.processAndSend(key, all, outputChannel, deleteOrTrackCallback());
} 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());
}
}
}
}
}
private class DelayedKey implements Delayed {
private Object key;
private Long releaseTime;
private TimeUnit unit = TimeUnit.MILLISECONDS;
public DelayedKey(Object correlationKey, long delay) {
Assert.notNull(correlationKey, "'correlationKey' must not be null");
this.key = correlationKey;
this.releaseTime = System.currentTimeMillis() + delay;
}
public long getDelay(TimeUnit unit) {
return unit.convert(this.releaseTime - System.currentTimeMillis(), this.unit);
}
public int compareTo(Delayed o) {
return ((Long) this.getDelay(this.unit)).compareTo(o.getDelay(this.unit));
}
public Object getKey() {
return key;
}
}
}

75
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/DefaultResequencerStrategies.java Normal file
View File

@@ -0,0 +1,75 @@
/*
* 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.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.core.MessageHeaders;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicInteger;
/**
* @author Iwein Fuld
*/
public class DefaultResequencerStrategies implements CorrelationStrategy, CompletionStrategy, MessagesProcessor {
private final ConcurrentMap<Object, AtomicInteger> nextMessagesToPass = new ConcurrentHashMap<Object, AtomicInteger>();
private volatile DefaultResequencerStrategies.SequenceNumberComparator sequenceSizeComparator = new SequenceNumberComparator();
private volatile boolean releasePartialSequences;
public Object getCorrelationKey(Message<?> message) {
Object key = message.getHeaders().getCorrelationId();
nextMessagesToPass.putIfAbsent(key, new AtomicInteger(1));
return key;
}
public boolean isComplete(List<Message<?>> messages) {
return releasePartialSequences||
messages.get(0).getHeaders().getSequenceSize()==messages.size();
}
public void processAndSend(Object correlationKey, Collection<Message<?>> all, MessageChannel outputChannel, BufferedMessagesCallback processedCallback) {
if (all.size() > 0) {
List<Message> sorted = new ArrayList(all);
Collections.sort(sorted, sequenceSizeComparator);
AtomicInteger nextSequence = nextMessagesToPass.get(correlationKey);
for (Message message : sorted) {
final int sequenceNumber = message.getHeaders().getSequenceNumber();
if (sequenceNumber <= nextSequence.get()) {
outputChannel.send(message);
nextSequence.compareAndSet(sequenceNumber, sequenceNumber + 1);
processedCallback.onProcessingOf(message);
}
}
MessageHeaders headers = sorted.get(0).getHeaders();
if (all.size() == headers.getSequenceSize()){
processedCallback.onCompletionOf(correlationKey);
}
}
}
public void setReleasePartialSequences(boolean releasePartialSequences) {
this.releasePartialSequences = releasePartialSequences;
}
private class SequenceNumberComparator implements Comparator<Message> {
public int compare(Message o1, Message o2) {
return o1.getHeaders().getSequenceNumber().compareTo(o2.getHeaders().getSequenceNumber());
}
}
}

18
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessagesProcessor.java Normal file
View File

@@ -0,0 +1,18 @@
package org.springframework.integration.aggregator;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.aggregator.BufferedMessagesCallback;
import java.util.Collection;
/**
* @author Iwein Fuld
*/
public interface MessagesProcessor {
void processAndSend(Object correlationKey,
Collection<Message<?>> messagesUpForProcessing,
MessageChannel outputChannel,
BufferedMessagesCallback processedCallback);
}

15
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/PassThroughMessagesProcessor.java Normal file
View File

@@ -0,0 +1,15 @@
package org.springframework.integration.aggregator;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import java.util.Collection;
public class PassThroughMessagesProcessor implements MessagesProcessor {
public void processAndSend(Object correlationKey, Collection<Message<?>> messagesUpForProcessing, MessageChannel outputChannel, BufferedMessagesCallback processedCallback) {
for (Message<?> message : messagesUpForProcessing) {
outputChannel.send(message);
}
}
}

45
org.springframework.integration/src/main/java/org/springframework/integration/store/MessageStore.java Normal file
View File

@@ -0,0 +1,45 @@
/*
* Copyright 2002-2008 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.store;
import java.util.List;
import org.springframework.integration.core.Message;
/**
* Strategy interface for storing and retrieving messages. The interface mimics
* the semantics for REST for the methods named after REST operations. This is
* helpful when mapping to a RESTful api.
*
* @author Mark Fisher
* @author Iwein Fuld
*/
public interface MessageStore {
Message<?> get(Object key);
<T> Message<T> put(Message<T> message);
<T> Message<T> post(T payload);
Message<?> delete(Object key);
List<Message<?>> list();
List<Message<?>> getAll(Object correlationKey);
}

86
org.springframework.integration/src/main/java/org/springframework/integration/store/SimpleMessageStore.java Normal file
View File

@@ -0,0 +1,86 @@
/*
* Copyright 2002-2008 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.store;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import org.springframework.integration.core.Message;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.util.Assert;
/**
* Map-based implementation of {@link MessageStore} that enforces a maximum capacity.
*
* @author Iwein Fuld
* @author Mark Fisher
*/
public class SimpleMessageStore implements MessageStore {
private final Map<Object, Message<?>> map;
public SimpleMessageStore(int capacity) {
this.map = new ConcurrentHashMap<Object, Message<?>>(capacity);
}
@SuppressWarnings("unchecked")
public <T> Message<T> put( Message<T> message) {
return (Message<T>) this.map.put(message.getHeaders().getId(), message);
}
public Message<?> get(Object key) {
return (key != null) ? this.map.get(key) : null;
}
public List<Message<?>> list() {
return new ArrayList<Message<?>>(this.map.values());
}
public Message<?> delete(Object key) {
return (key != null) ? this.map.remove(key) : null;
}
public int size() {
return this.map.size();
}
public List<Message<?>> getAll(Object correlationKey) {
Assert.notNull(correlationKey, "'correlationKey' must not be null");
List<Message<?>> matched = new ArrayList<Message<?>>();
Collection<Message<?>> values = map.values();
for (Message<?> message : values) {
if(message.getHeaders().getCorrelationId().equals(correlationKey)){
matched.add(message);
}
}
return matched;
}
public <T> Message<T> post(T payload) {
Message<T> message = MessageBuilder.withPayload(payload).build();
this.put(message);
return message;
}
}

74
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/BufferingMessageHandlerIntegrationTest.java Normal file
View File

@@ -0,0 +1,74 @@
package org.springframework.integration.aggregator;
import org.junit.Test;
import org.junit.Before;
import static org.mockito.Mockito.*;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.store.MessageStore;
import org.springframework.integration.store.SimpleMessageStore;
public class BufferingMessageHandlerIntegrationTest {
private CompletionStrategy completionStrategy;
private CorrelationStrategy correlationStrategy;
private MessageStore store = new SimpleMessageStore(100);
private MessageChannel outputChannel = mock(MessageChannel.class);
private MessagesProcessor processor = new PassThroughMessagesProcessor();
// private BufferingMessageHandler customizedHandler = new BufferingMessageHandler(
// store, correlationStrategy, completionStrategy, processor,
// outputChannel);
private BufferingMessageHandler defaultHandler = new BufferingMessageHandler(
store, processor);
@Before public void setupHandler(){
defaultHandler.setOutputChannel(outputChannel);
}
private Message<?> correlatedMessage(Object correlationId,
Integer sequenceSize, Integer sequenceNumber) {
return MessageBuilder.withPayload("test").setCorrelationId(
correlationId).setSequenceNumber(sequenceNumber)
.setSequenceSize(sequenceSize).build();
}
@Test
public void completesSingleMessage() throws Exception {
Message<?> message = correlatedMessage(1,
1, 1);
defaultHandler.handleMessage(message);
verify(outputChannel).send(message);
}
@Test
public void completesAfterSequenceComplete() throws Exception {
Message<?> message1 = correlatedMessage(1, 2, 1);
Message<?> message2 = correlatedMessage(1, 2, 2);
defaultHandler.handleMessage(message1);
verify(outputChannel, never()).send(message1);
defaultHandler.handleMessage(message2);
verify(outputChannel).send(message1);
verify(outputChannel).send(message2);
}
@Test
public void completesWithoutReleasingIncompleteCorrellations() throws Exception {
Message<?> message1 = correlatedMessage(1, 2, 1);
Message<?> message2 = correlatedMessage(2, 2, 2);
Message<?> message1a = correlatedMessage(1, 2, 1);
Message<?> message2a = correlatedMessage(2, 2, 2);
defaultHandler.handleMessage(message1);
defaultHandler.handleMessage(message2);
verify(outputChannel, never()).send(message1);
verify(outputChannel, never()).send(message2);
defaultHandler.handleMessage(message1a);
verify(outputChannel).send(message1);
verify(outputChannel).send(message1a);
verify(outputChannel, never()).send(message2);
verify(outputChannel, never()).send(message2a);
defaultHandler.handleMessage(message2a);
verify(outputChannel).send(message2);
verify(outputChannel).send(message2a);
}
}

94
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/BufferingMessageHandlerTest.java Normal file
View File

@@ -0,0 +1,94 @@
/*
* 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;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mock;
import static org.mockito.Mockito.*;
import org.mockito.runners.MockitoJUnit44Runner;
import org.springframework.integration.aggregator.CompletionStrategy;
import org.springframework.integration.aggregator.CorrelationStrategy;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.core.MessageHeaders;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.store.MessageStore;
import java.util.ArrayList;
import java.util.List;
/**
* @author Iwein Fuld
*/
@RunWith(MockitoJUnit44Runner.class)
public class BufferingMessageHandlerTest {
private BufferingMessageHandler buffer;
@Mock
private MessageStore store;
@Mock
private CorrelationStrategy correlationStrategy;
@Mock
private CompletionStrategy completionStrategy;
@Mock
private MessagesProcessor processor;
@Mock
private MessageChannel outputChannel;
@Before
public void initializeSubject() {
buffer = new BufferingMessageHandler(store, correlationStrategy,
completionStrategy, processor);
buffer.setOutputChannel(outputChannel);
}
@Test
public void bufferCompletesNormally() throws Exception {
String correlationKey = "key";
Message<?> message1 = testMessage(1);
Message<?> message2 = testMessage(2);
List<Message<?>> storedMessages = new ArrayList<Message<?>>();
when(correlationStrategy.getCorrelationKey(isA(Message.class)))
.thenReturn(correlationKey);
when(completionStrategy.isComplete(storedMessages)).thenReturn(false);
storedMessages.add(message1);
when(store.getAll(correlationKey)).thenReturn(storedMessages);
buffer.handleMessageInternal(message1);
storedMessages.add(message2);
when(store.getAll(correlationKey)).thenReturn(storedMessages);
when(completionStrategy.isComplete(storedMessages)).thenReturn(true);
buffer.handleMessageInternal(message2);
verify(store).put(message1);
verify(store).put(message2);
verify(store, times(2)).getAll(correlationKey);
verify(correlationStrategy).getCorrelationKey(message1);
verify(correlationStrategy).getCorrelationKey(message2);
verify(completionStrategy, times(2)).isComplete(storedMessages);
verify(processor).
processAndSend(eq(correlationKey), eq(storedMessages), eq(outputChannel), isA(BufferedMessagesCallback.class));
}
private Message<?> testMessage(int id) {
return MessageBuilder.withPayload("test").setHeader(MessageHeaders.ID,
id).build();
}
}

275
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/NewResequencerTests.java Normal file
View File

@@ -0,0 +1,275 @@
/*
* 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.After;
import static org.junit.Assert.*;
import static org.hamcrest.CoreMatchers.*;
import org.junit.Before;
import org.junit.Test;
import org.junit.Ignore;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.aggregator.BufferingMessageHandler;
import org.springframework.integration.store.MessageStore;
import org.springframework.integration.store.SimpleMessageStore;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
/**
* @author Marius Bogoevici
* @author Alex Peters
* @author Iwein Fuld
*/
public class NewResequencerTests {
private BufferingMessageHandler resequencer;
private ThreadPoolTaskScheduler taskScheduler;
private DefaultResequencerStrategies resequencerStrategies;
@Before
public void configureResequencer() {
this.resequencerStrategies = new DefaultResequencerStrategies();
MessageStore store = new SimpleMessageStore(30);
this.resequencer = new BufferingMessageHandler(store, resequencerStrategies, resequencerStrategies, resequencerStrategies);
this.taskScheduler = TestUtils.createTaskScheduler(10);
this.resequencer.setTaskScheduler(taskScheduler);
this.taskScheduler.afterPropertiesSet();
this.resequencer.start();
}
@Test
public void testBasicResequencing() throws InterruptedException {
QueueChannel replyChannel = new QueueChannel();
Message<?> message1 = createMessage("123", "ABC", 3, 3, replyChannel);
Message<?> message2 = createMessage("456", "ABC", 3, 1, replyChannel);
Message<?> message3 = createMessage("789", "ABC", 3, 2, replyChannel);
this.resequencer.handleMessage(message1);
this.resequencer.handleMessage(message3);
this.resequencer.handleMessage(message2);
Message<?> reply1 = replyChannel.receive(0);
Message<?> reply2 = replyChannel.receive(0);
Message<?> reply3 = replyChannel.receive(0);
assertNotNull(reply1);
assertEquals(new Integer(1), reply1.getHeaders().getSequenceNumber());
assertNotNull(reply2);
assertEquals(new Integer(2), reply2.getHeaders().getSequenceNumber());
assertNotNull(reply3);
assertEquals(new Integer(3), reply3.getHeaders().getSequenceNumber());
}
@Test
public void testResequencingWithDuplicateMessages() {
QueueChannel replyChannel = new QueueChannel();
Message<?> message1 = createMessage("123", "ABC", 3, 3, replyChannel);
Message<?> message2 = createMessage("456", "ABC", 3, 1, replyChannel);
Message<?> message3 = createMessage("789", "ABC", 3, 2, replyChannel);
this.resequencer.handleMessage(message1);
this.resequencer.handleMessage(message3);
this.resequencer.handleMessage(message3);
this.resequencer.handleMessage(message2);
Message<?> reply1 = replyChannel.receive(0);
Message<?> reply2 = replyChannel.receive(0);
Message<?> reply3 = replyChannel.receive(0);
assertNotNull(reply1);
assertEquals(new Integer(1), reply1.getHeaders().getSequenceNumber());
assertNotNull(reply2);
assertEquals(new Integer(2), reply2.getHeaders().getSequenceNumber());
assertNotNull(reply3);
assertEquals(new Integer(3), reply3.getHeaders().getSequenceNumber());
}
@Test
public void testResequencingWithIncompleteSequenceRelease() throws InterruptedException {
this.resequencerStrategies.setReleasePartialSequences(true);
QueueChannel replyChannel = new QueueChannel();
Message<?> message1 = createMessage("123", "ABC", 4, 2, replyChannel);
Message<?> message2 = createMessage("456", "ABC", 4, 1, replyChannel);
Message<?> message3 = createMessage("789", "ABC", 4, 4, replyChannel);
Message<?> message4 = createMessage("XYZ", "ABC", 4, 3, replyChannel);
this.resequencer.handleMessage(message1);
this.resequencer.handleMessage(message2);
this.resequencer.handleMessage(message3);
Message<?> reply1 = replyChannel.receive(0);
Message<?> reply2 = replyChannel.receive(0);
Message<?> reply3 = replyChannel.receive(0);
// only messages 1 and 2 should have been received by now
assertNotNull(reply1);
assertEquals(new Integer(1), reply1.getHeaders().getSequenceNumber());
assertNotNull(reply2);
assertEquals(new Integer(2), reply2.getHeaders().getSequenceNumber());
assertNull(reply3);
// when sending the last message, the whole sequence must have been sent
this.resequencer.handleMessage(message4);
reply3 = replyChannel.receive(0);
Message<?> reply4 = replyChannel.receive(0);
assertNotNull(reply3);
assertEquals(new Integer(3), reply3.getHeaders().getSequenceNumber());
assertNotNull(reply4);
assertEquals(new Integer(4), reply4.getHeaders().getSequenceNumber());
}
@Test
public void testResequencingWithDiscard() throws InterruptedException {
QueueChannel replyChannel = new QueueChannel();
QueueChannel discardChannel = new QueueChannel();
Message<?> message1 = createMessage("123", "ABC", 3, 2, replyChannel);
Message<?> message2 = createMessage("456", "ABC", 3, 1, replyChannel);
Message<?> message3 = createMessage("789", "ABC", 3, 3, replyChannel);
this.resequencer.setSendPartialResultOnTimeout(false);
this.resequencerStrategies.setReleasePartialSequences(false);
this.resequencer.setDiscardChannel(discardChannel);
this.resequencer.setTimeout(90000);
this.resequencer.handleMessage(message1);
this.resequencer.handleMessage(message2);
//this.resequencer.discardBarrier(this.resequencer.barriers.get("ABC"));
this.resequencer.forceComplete("ABC");
Message<?> reply1 = discardChannel.receive(0);
Message<?> reply2 = discardChannel.receive(0);
Message<?> reply3 = discardChannel.receive(0);
// only messages 1 and 2 should have been received by now
// messages need not be reordered
assertNotNull(reply1);
assertThat( reply1.getHeaders().getSequenceNumber(), is(new Integer(2)));
assertNotNull(reply2);
assertThat( reply2.getHeaders().getSequenceNumber(), is(new Integer(1)));
assertNull(reply3);
// when sending the last message, it waits in the buffer for retries of the other two
this.resequencer.handleMessage(message3);
reply3 = discardChannel.receive(0);
assertNull(reply3);
this.resequencer.handleMessage(message1);
this.resequencer.handleMessage(message2);
reply1 = replyChannel.receive(0);
reply2 = replyChannel.receive(0);
reply3 = replyChannel.receive(0);
assertNotNull(reply1);
assertThat( reply1.getHeaders().getSequenceNumber(), is(new Integer(1)));
assertNotNull(reply2);
assertThat( reply2.getHeaders().getSequenceNumber(), is(new Integer(2)));
assertNotNull(reply3);
assertThat( reply3.getHeaders().getSequenceNumber(), is(new Integer(3)));
}
@Test
@Ignore //different sequence sizes are not supported
public void testResequencingWithDifferentSequenceSizes() throws InterruptedException {
QueueChannel replyChannel = new QueueChannel();
QueueChannel discardChannel = new QueueChannel();
Message<?> message1 = createMessage("123", "ABC", 4, 2, replyChannel);
Message<?> message2 = createMessage("456", "ABC", 5, 1, replyChannel);
this.resequencer.setSendPartialResultOnTimeout(false);
//this.resequencer.setReleasePartialSequences(false);
this.resequencer.setDiscardChannel(discardChannel);
this.resequencer.setTimeout(90000);
this.resequencer.handleMessage(message1);
this.resequencer.handleMessage(message2);
//this.resequencer.discardBarrier(this.resequencer.barriers.get("ABC"));
Message<?> reply1 = discardChannel.receive(0);
Message<?> reply2 = discardChannel.receive(0);
// only messages 1 - with sequence number 2 - should have been received by now
// the other has been discarded
assertNotNull(reply1);
assertEquals(new Integer(2), reply1.getHeaders().getSequenceNumber());
assertNull(reply2);
}
@Test
public void testResequencingWithWrongSequenceSizeAndNumber() throws InterruptedException {
QueueChannel replyChannel = new QueueChannel();
QueueChannel discardChannel = new QueueChannel();
Message<?> message1 = createMessage("123", "ABC", 2, 4, replyChannel);
this.resequencer.setSendPartialResultOnTimeout(false);
//this.resequencer.setReleasePartialSequences(false);
this.resequencer.setDiscardChannel(discardChannel);
this.resequencer.setTimeout(90000);
this.resequencer.handleMessage(message1);
//this.resequencer.discardBarrier(this.resequencer.barriers.get("ABC"));
Message<?> reply1 = discardChannel.receive(0);
// No message has been received - the message has been rejected.
assertNull(reply1);
}
@Test
public void testResequencingWithCompleteSequenceRelease() throws InterruptedException {
//this.resequencer.setReleasePartialSequences(false);
QueueChannel replyChannel = new QueueChannel();
Message<?> message1 = createMessage("123", "ABC", 4, 2, replyChannel);
Message<?> message2 = createMessage("456", "ABC", 4, 1, replyChannel);
Message<?> message3 = createMessage("789", "ABC", 4, 4, replyChannel);
Message<?> message4 = createMessage("XYZ", "ABC", 4, 3, replyChannel);
this.resequencer.handleMessage(message1);
this.resequencer.handleMessage(message2);
this.resequencer.handleMessage(message3);
Message<?> reply1 = replyChannel.receive(0);
Message<?> reply2 = replyChannel.receive(0);
Message<?> reply3 = replyChannel.receive(0);
// no messages should have been received yet
assertNull(reply1);
assertNull(reply2);
assertNull(reply3);
// after sending the last message, the whole sequence should have been sent
this.resequencer.handleMessage(message4);
reply1 = replyChannel.receive(0);
reply2 = replyChannel.receive(0);
reply3 = replyChannel.receive(0);
Message<?> reply4 = replyChannel.receive(0);
assertNotNull(reply1);
assertEquals(new Integer(1), reply1.getHeaders().getSequenceNumber());
assertNotNull(reply2);
assertEquals(new Integer(2), reply2.getHeaders().getSequenceNumber());
assertNotNull(reply3);
assertEquals(new Integer(3), reply3.getHeaders().getSequenceNumber());
assertNotNull(reply4);
assertEquals(new Integer(4), reply4.getHeaders().getSequenceNumber());
}
@Test
public void testRemovalOfBarrierWhenLastMessageOfSequenceArrives() {
QueueChannel replyChannel = new QueueChannel();
String correlationId = "ABC";
Message<?> message1 = createMessage("123", correlationId, 1, 1,
replyChannel);
resequencer.handleMessage(message1);
//assertThat(resequencer.barriers.containsKey(correlationId), is(false));
}
private static Message<?> createMessage(String payload, Object correlationId,
int sequenceSize, int sequenceNumber, MessageChannel replyChannel) {
return MessageBuilder.withPayload(payload)
.setCorrelationId(correlationId)
.setSequenceSize(sequenceSize)
.setSequenceNumber(sequenceNumber)
.setReplyChannel(replyChannel)
.build();
}
@After
public void stopTaskScheduler() {
this.resequencer.stop();
this.taskScheduler.destroy();
}
}

2
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/integration/ConcurrentAggregatorIntegrationTests-context.xml
View File

@@ -6,8 +6,6 @@
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration.xsd">
<annotation-config />
<channel id="input">
<queue capacity="5" />
</channel>

10
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/integration/ConcurrentAggregatorIntegrationTests.java
View File

@@ -19,11 +19,6 @@ package org.springframework.integration.aggregator.integration;
import static org.hamcrest.CoreMatchers.is;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertThat;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
@@ -37,6 +32,10 @@ import org.springframework.integration.message.GenericMessage;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
/**
* @author Iwein Fuld
* @author Alex Peters
@@ -87,7 +86,6 @@ public class ConcurrentAggregatorIntegrationTests {
headers.put(MessageHeaders.SEQUENCE_NUMBER, sequenceNumber);
headers.put(MessageHeaders.SEQUENCE_SIZE, sequenceSize);
headers.put(MessageHeaders.CORRELATION_ID, correllationId);
headers.put(MessageHeaders.ID, 1);
return headers;
}

360
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/integration/NewAggregatorEndpointTests.java Normal file
View File

@@ -0,0 +1,360 @@
/*
* 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.integration;
import org.junit.After;
import static org.junit.Assert.*;
import static org.hamcrest.CoreMatchers.*;
import org.junit.Before;
import org.junit.Test;
import org.junit.Ignore;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.core.MessageHeaders;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.message.MessageHandler;
import org.springframework.integration.message.MessageHandlingException;
import org.springframework.integration.aggregator.BufferingMessageHandler;
import org.springframework.integration.aggregator.MessagesProcessor;
import org.springframework.integration.aggregator.BufferedMessagesCallback;
import org.springframework.integration.store.SimpleMessageStore;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import java.util.Collection;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
/**
* @author Mark Fisher
* @author Marius Bogoevici
* @author Iwein Fuld
*/
public class NewAggregatorEndpointTests {
private TaskExecutor taskExecutor;
private ThreadPoolTaskScheduler taskScheduler;
private BufferingMessageHandler aggregator;
@Before
public void configureAggregator() {
this.taskExecutor = new SimpleAsyncTaskExecutor();
this.taskScheduler = new ThreadPoolTaskScheduler();
taskScheduler.afterPropertiesSet();
this.taskScheduler.afterPropertiesSet();
this.aggregator = new BufferingMessageHandler(new SimpleMessageStore(50), new MultiplyingProcessor());
this.aggregator.setTaskScheduler(this.taskScheduler);
}
@Test
public void testCompleteGroupWithinTimeout() throws InterruptedException {
this.aggregator.start();
QueueChannel replyChannel = new QueueChannel();
Message<?> message1 = createMessage(3, "ABC", 3, 1, replyChannel, null);
Message<?> message2 = createMessage(5, "ABC", 3, 2, replyChannel, null);
Message<?> message3 = createMessage(7, "ABC", 3, 3, replyChannel, null);
CountDownLatch latch = new CountDownLatch(3);
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);
Message<?> reply = replyChannel.receive(2000);
assertNotNull(reply);
assertEquals(reply.getPayload(), 105);
}
@Test
@Ignore
//dropped backwards compatibility for duplicate ID's
public void testCompleteGroupWithinTimeoutWithSameId() throws InterruptedException {
this.aggregator.start();
QueueChannel replyChannel = new QueueChannel();
Message<?> message1 = createMessage(3, "ABC", 3, 1, replyChannel, "ID#1");
Message<?> message2 = createMessage(5, "ABC", 3, 2, replyChannel, "ID#1");
Message<?> message3 = createMessage(7, "ABC", 3, 3, replyChannel, "ID#1");
CountDownLatch latch = new CountDownLatch(3);
//for testing the duplication scenario, the messages must be processed synchronously
new AggregatorTestTask(this.aggregator, message1, latch).run();
new AggregatorTestTask(this.aggregator, message2, latch).run();
new AggregatorTestTask(this.aggregator, message3, latch).run();
Message<?> reply = replyChannel.receive(500);
assertNotNull(reply);
assertEquals("123456789", reply.getPayload());
}
@Test
public void testShouldNotSendPartialResultOnTimeoutByDefault() throws InterruptedException {
this.aggregator.start();
QueueChannel discardChannel = new QueueChannel();
this.aggregator.setTimeout(50);
this.aggregator.setReaperInterval(10);
this.aggregator.setDiscardChannel(discardChannel);
QueueChannel replyChannel = new QueueChannel();
Message<?> message = createMessage(3, "ABC", 2, 1, replyChannel, null);
CountDownLatch latch = new CountDownLatch(1);
AggregatorTestTask task = new AggregatorTestTask(this.aggregator, message, latch);
this.taskExecutor.execute(task);
latch.await(2000, TimeUnit.MILLISECONDS);
assertEquals("Task should have completed within timeout", 0, latch.getCount());
Message<?> reply = replyChannel.receive(100);
assertNull("No message should have been sent normally", reply);
Message<?> discardedMessage = discardChannel.receive(100);
assertNotNull("A message should have been discarded", discardedMessage);
assertEquals(message, discardedMessage);
}
@Test
public void testShouldSendPartialResultOnTimeoutTrue() throws InterruptedException {
this.aggregator.setTimeout(500);
this.aggregator.setReaperInterval(10);
this.aggregator.setSendPartialResultOnTimeout(true);
this.aggregator.start();
QueueChannel replyChannel = new QueueChannel();
Message<?> message1 = createMessage(3, "ABC", 3, 1, replyChannel, null);
Message<?> message2 = createMessage(5, "ABC", 3, 2, replyChannel, null);
CountDownLatch latch = new CountDownLatch(2);
AggregatorTestTask task1 = new AggregatorTestTask(this.aggregator, message1, latch);
AggregatorTestTask task2 = new AggregatorTestTask(this.aggregator, message2, latch);
this.taskExecutor.execute(task1);
this.taskExecutor.execute(task2);
latch.await(3000, TimeUnit.MILLISECONDS);
assertEquals("handlers should have been invoked within time limit", 0, latch.getCount());
Message<?> reply = replyChannel.receive(3000);
assertNotNull("A reply message should have been received", reply);
assertEquals(15, reply.getPayload());
assertNull(task1.getException());
assertNull(task2.getException());
}
@Test
public void testMultipleGroupsSimultaneously() throws InterruptedException {
this.aggregator.start();
QueueChannel replyChannel1 = new QueueChannel();
QueueChannel replyChannel2 = new QueueChannel();
Message<?> message1 = createMessage(3, "ABC", 3, 1, replyChannel1, null);
Message<?> message2 = createMessage(5, "ABC", 3, 2, replyChannel1, null);
Message<?> message3 = createMessage(7, "ABC", 3, 3, replyChannel1, null);
Message<?> message4 = createMessage(11, "XYZ", 3, 1, replyChannel2, null);
Message<?> message5 = createMessage(13, "XYZ", 3, 2, replyChannel2, null);
Message<?> message6 = createMessage(17, "XYZ", 3, 3, replyChannel2, null);
CountDownLatch latch = new CountDownLatch(6);
this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, message1, latch));
this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, message6, latch));
this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, message2, latch));
this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, message5, latch));
this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, message3, latch));
this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, message4, latch));
latch.await(1000, TimeUnit.MILLISECONDS);
Message<Integer> reply1 = (Message<Integer>) replyChannel1.receive(500);
assertNotNull(reply1);
assertThat(reply1.getPayload(), is(105));
Message<Integer> reply2 = (Message<Integer>) replyChannel2.receive(500);
assertNotNull(reply2);
assertThat(reply2.getPayload(), is(2431));
}
@Test
public void testDiscardChannelForTrackedCorrelationId() {
this.aggregator.start();
QueueChannel replyChannel = new QueueChannel();
QueueChannel discardChannel = new QueueChannel();
this.aggregator.setDiscardChannel(discardChannel);
this.aggregator.handleMessage(createMessage(1, "tracked", 1, 1, replyChannel, null));
Message<?> received1 = replyChannel.receive(100);
assertEquals(1, received1.getPayload());
assertNotNull("Expected aggregated message, but got null", received1);
this.aggregator.handleMessage(createMessage(2, "tracked", 1, 1, replyChannel, null));
Message<?> received2 = discardChannel.receive(1000);
assertNotNull("Expected discarded message, but got null", received2);
assertEquals(2, received2.getPayload());
}
@Test
@Ignore
//dropped backwards compatibility for setting capacity limit (it's always Integer.MAX_VALUE)
public void testTrackedCorrelationIdsCapacityAtLimit() {
this.aggregator.start();
QueueChannel replyChannel = new QueueChannel();
QueueChannel discardChannel = new QueueChannel();
//this.aggregator.setTrackedCorrelationIdCapacity(3);
this.aggregator.setDiscardChannel(discardChannel);
this.aggregator.handleMessage(createMessage(1, 1, 1, 1, replyChannel, null));
assertEquals(1, replyChannel.receive(100).getPayload());
this.aggregator.handleMessage(createMessage(3, 2, 1, 1, replyChannel, null));
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
this.aggregator.handleMessage(createMessage(2, 1, 1, 1, replyChannel, null));
assertEquals(2, discardChannel.receive(100).getPayload());
}
@Test
@Ignore
//dropped backwards compatibility for setting capacity limit (it's always Integer.MAX_VALUE)
public void testTrackedCorrelationIdsCapacityPassesLimit() {
this.aggregator.start();
QueueChannel replyChannel = new QueueChannel();
QueueChannel discardChannel = new QueueChannel();
//this.aggregator.setTrackedCorrelationIdCapacity(3);
this.aggregator.setDiscardChannel(discardChannel);
this.aggregator.handleMessage(createMessage(1, 1, 1, 1, replyChannel, null));
assertEquals(1, replyChannel.receive(100).getPayload());
this.aggregator.handleMessage(createMessage(2, 2, 1, 1, replyChannel, null));
assertEquals(2, replyChannel.receive(100).getPayload());
this.aggregator.handleMessage(createMessage(3, 3, 1, 1, replyChannel, null));
assertEquals(3, replyChannel.receive(100).getPayload());
this.aggregator.handleMessage(createMessage(4, 4, 1, 1, replyChannel, null));
assertEquals(4, replyChannel.receive(100).getPayload());
this.aggregator.handleMessage(createMessage(5, 1, 1, 1, replyChannel, null));
assertEquals(5, replyChannel.receive(100).getPayload());
assertNull(discardChannel.receive(0));
}
@Test(expected = MessageHandlingException.class)
public void testExceptionThrownIfNoCorrelationId() throws InterruptedException {
this.aggregator.start();
Message<?> message = createMessage(3, null, 2, 1, new QueueChannel(), null);
this.aggregator.handleMessage(message);
}
@Test
public void testAdditionalMessageAfterCompletion() throws InterruptedException {
this.aggregator.start();
QueueChannel replyChannel = new QueueChannel();
Message<?> message1 = createMessage(3, "ABC", 3, 1, replyChannel, null);
Message<?> message2 = createMessage(5, "ABC", 3, 2, replyChannel, null);
Message<?> message3 = createMessage(7, "ABC", 3, 3, replyChannel, null);
Message<?> message4 = createMessage(33, "ABC", 3, 3, replyChannel, null);
CountDownLatch latch = new CountDownLatch(4);
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));
this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, message4, latch));
latch.await(1000, TimeUnit.MILLISECONDS);
Message<?> reply = replyChannel.receive(500);
assertNotNull("A message should be aggregated", reply);
assertThat(((Integer) reply.getPayload()), is(105));
}
@Test
public void testNullReturningAggregator() throws InterruptedException {
this.aggregator.start();
this.aggregator = new BufferingMessageHandler(new SimpleMessageStore(50), new NullReturningMessageProcessor());
this.aggregator.setTaskScheduler(this.taskScheduler);
QueueChannel replyChannel = new QueueChannel();
Message<?> message1 = createMessage(3, "ABC", 3, 1, replyChannel, null);
Message<?> message2 = createMessage(5, "ABC", 3, 2, replyChannel, null);
Message<?> message3 = createMessage(7, "ABC", 3, 3, replyChannel, null);
CountDownLatch latch = new CountDownLatch(3);
AggregatorTestTask task1 = new AggregatorTestTask(aggregator, message1, latch);
this.taskExecutor.execute(task1);
AggregatorTestTask task2 = new AggregatorTestTask(aggregator, message2, latch);
this.taskExecutor.execute(task2);
AggregatorTestTask task3 = new AggregatorTestTask(aggregator, message3, latch);
this.taskExecutor.execute(task3);
latch.await(1000, TimeUnit.MILLISECONDS);
assertNull(task1.getException());
assertNull(task2.getException());
assertNull(task3.getException());
Message<?> reply = replyChannel.receive(500);
assertNull(reply);
}
private static Message<?> createMessage(Object payload, Object correlationId,
int sequenceSize, int sequenceNumber, MessageChannel replyChannel, String predefinedId) {
MessageBuilder<Object> builder = MessageBuilder.withPayload(payload)
.setCorrelationId(correlationId)
.setSequenceSize(sequenceSize)
.setSequenceNumber(sequenceNumber)
.setReplyChannel(replyChannel);
if (predefinedId != null) {
builder.setHeader(MessageHeaders.ID, predefinedId);
}
return builder.build();
}
private static class AggregatorTestTask implements Runnable {
private MessageHandler aggregator;
private Message<?> message;
private Exception exception;
private CountDownLatch latch;
AggregatorTestTask(MessageHandler aggregator, Message<?> message, CountDownLatch latch) {
this.aggregator = aggregator;
this.message = message;
this.latch = latch;
}
public Exception getException() {
return this.exception;
}
public void run() {
try {
this.aggregator.handleMessage(message);
}
catch (Exception e) {
e.printStackTrace();
this.exception = e;
}
finally {
this.latch.countDown();
}
}
}
@After
public void stopTaskScheduler() {
if (this.taskScheduler != null) this.taskScheduler.destroy();
if (this.aggregator != null) this.aggregator.stop();
}
private class MultiplyingProcessor implements MessagesProcessor {
public void processAndSend(Object correlationKey, Collection<Message<?>> messagesUpForProcessing,
MessageChannel outputChannel, BufferedMessagesCallback processedCallback
) {
Integer product = 1;
for (Message<?> message : messagesUpForProcessing) {
product *= (Integer) message.getPayload();
}
outputChannel.send(MessageBuilder.withPayload(product).build());
processedCallback.onProcessingOf(
messagesUpForProcessing.toArray(new Message[messagesUpForProcessing.size()])
);
processedCallback.onCompletionOf(correlationKey);
}
}
private class NullReturningMessageProcessor implements MessagesProcessor {
public void processAndSend(Object correlationKey, Collection<Message<?>> messagesUpForProcessing, MessageChannel outputChannel, BufferedMessagesCallback processedCallback) {
//noop
}
}
}

159
org.springframework.integration/src/test/java/org/springframework/integration/test/util/TestUtils.java Normal file
View File

@@ -0,0 +1,159 @@
/*
* 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.test.util;
import static org.hamcrest.CoreMatchers.is;
import org.hamcrest.Matcher;
import static org.junit.Assert.assertThat;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.FatalBeanException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.context.support.GenericApplicationContext;
import org.springframework.integration.channel.BeanFactoryChannelResolver;
import org.springframework.integration.channel.MessagePublishingErrorHandler;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.endpoint.AbstractPollingEndpoint;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.MessageHandler;
import org.springframework.integration.message.MessageHandlingException;
import org.springframework.integration.message.MessageRejectedException;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.scheduling.support.ErrorHandler;
import org.springframework.scheduling.support.PeriodicTrigger;
import org.springframework.util.Assert;
import java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy;
/**
* @author Mark Fisher
* @author Iwein Fuld
*/
public abstract class TestUtils {
public static Object getPropertyValue(Object root, String propertyPath) {
Object value = null;
DirectFieldAccessor accessor = new DirectFieldAccessor(root);
String[] tokens = propertyPath.split("\\.");
for (int i = 0; i < tokens.length; i++) {
value = accessor.getPropertyValue(tokens[i]);
if (value != null) {
accessor = new DirectFieldAccessor(value);
} else if (i == tokens.length - 1) {
return null;
} else {
throw new IllegalArgumentException(
"intermediate property '" + tokens[i] + "' is null");
}
}
return value;
}
@SuppressWarnings("unchecked")
public static <T> T getPropertyValue(Object root, String propertyPath, Class<T> type) {
Object value = getPropertyValue(root, propertyPath);
Assert.isAssignable(type, value.getClass());
return (T) value;
}
public static TestApplicationContext createTestApplicationContext() {
TestApplicationContext context = new TestApplicationContext();
ErrorHandler errorHandler = new MessagePublishingErrorHandler(new BeanFactoryChannelResolver(context));
ThreadPoolTaskScheduler scheduler = createTaskScheduler(10);
scheduler.setErrorHandler(errorHandler);
registerBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME, scheduler, context);
return context;
}
public static ThreadPoolTaskScheduler createTaskScheduler(int poolSize) {
ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
scheduler.setPoolSize(poolSize);
scheduler.setRejectedExecutionHandler(new CallerRunsPolicy());
scheduler.afterPropertiesSet();
return scheduler;
}
private static void registerBean(String beanName, Object bean, BeanFactory beanFactory) {
Assert.notNull(beanName, "bean name must not be null");
ConfigurableListableBeanFactory configurableListableBeanFactory = null;
if (beanFactory instanceof ConfigurableListableBeanFactory) {
configurableListableBeanFactory = (ConfigurableListableBeanFactory) beanFactory;
} else if (beanFactory instanceof GenericApplicationContext) {
configurableListableBeanFactory = ((GenericApplicationContext) beanFactory).getBeanFactory();
}
if (bean instanceof BeanNameAware) {
((BeanNameAware) bean).setBeanName(beanName);
}
if (bean instanceof BeanFactoryAware) {
((BeanFactoryAware) bean).setBeanFactory(beanFactory);
}
if (bean instanceof InitializingBean) {
try {
((InitializingBean) bean).afterPropertiesSet();
}
catch (Exception e) {
throw new FatalBeanException("failed to register bean with test context", e);
}
}
configurableListableBeanFactory.registerSingleton(beanName, bean);
}
public static class TestApplicationContext extends GenericApplicationContext {
private TestApplicationContext() {
super();
}
public void registerChannel(String channelName, MessageChannel channel) {
if (channel.getName() != null) {
if (channelName == null) {
Assert.notNull(channel.getName(), "channel name must not be null");
channelName = channel.getName();
} else {
Assert.isTrue(channel.getName().equals(channelName),
"channel name has already been set with a conflicting value");
}
}
registerBean(channelName, channel, this);
}
public void registerEndpoint(String endpointName, AbstractEndpoint endpoint) {
if (endpoint instanceof AbstractPollingEndpoint) {
DirectFieldAccessor accessor = new DirectFieldAccessor(endpoint);
if (accessor.getPropertyValue("trigger") == null) {
((AbstractPollingEndpoint) endpoint).setTrigger(new PeriodicTrigger(10));
}
}
registerBean(endpointName, endpoint, this);
}
}
public static MessageHandler handlerExpecting(final Matcher<Message> messageMatcher) {
return new MessageHandler() {
public void handleMessage(Message<?> message) throws MessageRejectedException, MessageHandlingException, MessageDeliveryException {
assertThat(message, is(messageMatcher));
}
};
}
}