diff --git a/org.springframework.integration.adapter/.classpath b/org.springframework.integration.adapter/.classpath index 72afa3ce4c..a355be7fdc 100644 --- a/org.springframework.integration.adapter/.classpath +++ b/org.springframework.integration.adapter/.classpath @@ -5,13 +5,13 @@ - - - - - - - + + + + + + + diff --git a/org.springframework.integration.event/.classpath b/org.springframework.integration.event/.classpath index 72afa3ce4c..a355be7fdc 100644 --- a/org.springframework.integration.event/.classpath +++ b/org.springframework.integration.event/.classpath @@ -5,13 +5,13 @@ - - - - - - - + + + + + + + diff --git a/org.springframework.integration.file/.classpath b/org.springframework.integration.file/.classpath index 95d6d53dcd..f7b0324893 100644 --- a/org.springframework.integration.file/.classpath +++ b/org.springframework.integration.file/.classpath @@ -5,19 +5,19 @@ - - - - - - - - - - - - - + + + + + + + + + + + + + diff --git a/org.springframework.integration.http/.classpath b/org.springframework.integration.http/.classpath index a6a7a8b193..ed7669b16d 100644 --- a/org.springframework.integration.http/.classpath +++ b/org.springframework.integration.http/.classpath @@ -5,22 +5,22 @@ - - - - - - - - - - - - - - - - + + + + + + + + + + + + + + + + diff --git a/org.springframework.integration.httpinvoker/.classpath b/org.springframework.integration.httpinvoker/.classpath index 86b6c5aa14..7355875d8a 100644 --- a/org.springframework.integration.httpinvoker/.classpath +++ b/org.springframework.integration.httpinvoker/.classpath @@ -5,18 +5,18 @@ - - - - - - - - - - - - + + + + + + + + + + + + diff --git a/org.springframework.integration.jms/.classpath b/org.springframework.integration.jms/.classpath index abbed6ec96..89f640534b 100644 --- a/org.springframework.integration.jms/.classpath +++ b/org.springframework.integration.jms/.classpath @@ -5,17 +5,17 @@ - - - - - - - - - - - + + + + + + + + + + + diff --git a/org.springframework.integration.mail/.classpath b/org.springframework.integration.mail/.classpath index 35c0a4c170..755b0cbfb6 100644 --- a/org.springframework.integration.mail/.classpath +++ b/org.springframework.integration.mail/.classpath @@ -5,20 +5,20 @@ - - - - - - - - - - - - - - + + + + + + + + + + + + + + diff --git a/org.springframework.integration.rmi/.classpath b/org.springframework.integration.rmi/.classpath index b936e3abfe..1cfd4ef38c 100644 --- a/org.springframework.integration.rmi/.classpath +++ b/org.springframework.integration.rmi/.classpath @@ -5,16 +5,16 @@ - - - - - - - - - - + + + + + + + + + + diff --git a/org.springframework.integration.security/.classpath b/org.springframework.integration.security/.classpath index 80c53673a3..8fdbf22e09 100644 --- a/org.springframework.integration.security/.classpath +++ b/org.springframework.integration.security/.classpath @@ -5,19 +5,19 @@ - - - - - - - - - - - - - + + + + + + + + + + + + + diff --git a/org.springframework.integration.stream/.classpath b/org.springframework.integration.stream/.classpath index 6bc64653d2..fd04a0ed11 100644 --- a/org.springframework.integration.stream/.classpath +++ b/org.springframework.integration.stream/.classpath @@ -5,14 +5,14 @@ - - - - - - - - + + + + + + + + diff --git a/org.springframework.integration.test/.classpath b/org.springframework.integration.test/.classpath index d0ad05d0e4..33caaa3dee 100644 --- a/org.springframework.integration.test/.classpath +++ b/org.springframework.integration.test/.classpath @@ -5,21 +5,21 @@ - - - - - - - - - - - - - - - + + + + + + + + + + + + + + + diff --git a/org.springframework.integration/.classpath b/org.springframework.integration/.classpath index a8a1b828b4..e775664196 100644 --- a/org.springframework.integration/.classpath +++ b/org.springframework.integration/.classpath @@ -5,24 +5,23 @@ - - - - - - - - - - - - - - - - - - + + + + + + + + + + + + + + + + + diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/BufferedMessagesCallback.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/BufferedMessagesCallback.java new file mode 100644 index 0000000000..84d83b7047 --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/BufferedMessagesCallback.java @@ -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); +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/BufferingMessageHandler.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/BufferingMessageHandler.java new file mode 100644 index 0000000000..9e55e2d0e0 --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/BufferingMessageHandler.java @@ -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 keysInBuffer = new DelayQueue(); + 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> all = store.getAll(correlationKey); + complete(correlationKey, all, this.resolveReplyChannel(message)); + } else { + discardChannel.send(message); + } + } + + private boolean complete(Object correlationKey, List> 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> 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; + } + } +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/DefaultResequencerStrategies.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/DefaultResequencerStrategies.java new file mode 100644 index 0000000000..8aa80e5c9d --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/DefaultResequencerStrategies.java @@ -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 nextMessagesToPass = new ConcurrentHashMap(); + 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> messages) { + return releasePartialSequences|| + messages.get(0).getHeaders().getSequenceSize()==messages.size(); + } + + public void processAndSend(Object correlationKey, Collection> all, MessageChannel outputChannel, BufferedMessagesCallback processedCallback) { + if (all.size() > 0) { + List 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 { + public int compare(Message o1, Message o2) { + return o1.getHeaders().getSequenceNumber().compareTo(o2.getHeaders().getSequenceNumber()); + } + } +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessagesProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessagesProcessor.java new file mode 100644 index 0000000000..90a49a9b98 --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessagesProcessor.java @@ -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> messagesUpForProcessing, + MessageChannel outputChannel, + BufferedMessagesCallback processedCallback); +} \ No newline at end of file diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/PassThroughMessagesProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/PassThroughMessagesProcessor.java new file mode 100644 index 0000000000..244764e941 --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/PassThroughMessagesProcessor.java @@ -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> messagesUpForProcessing, MessageChannel outputChannel, BufferedMessagesCallback processedCallback) { + for (Message message : messagesUpForProcessing) { + outputChannel.send(message); + } + } +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/store/MessageStore.java b/org.springframework.integration/src/main/java/org/springframework/integration/store/MessageStore.java new file mode 100644 index 0000000000..7eeb5bcf1d --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/store/MessageStore.java @@ -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); + + Message put(Message message); + + Message post(T payload); + + Message delete(Object key); + + List> list(); + + List> getAll(Object correlationKey); + +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/store/SimpleMessageStore.java b/org.springframework.integration/src/main/java/org/springframework/integration/store/SimpleMessageStore.java new file mode 100644 index 0000000000..75c18cfd0d --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/store/SimpleMessageStore.java @@ -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> map; + + + public SimpleMessageStore(int capacity) { + this.map = new ConcurrentHashMap>(capacity); + } + + + @SuppressWarnings("unchecked") + public Message put( Message message) { + return (Message) this.map.put(message.getHeaders().getId(), message); + } + + public Message get(Object key) { + return (key != null) ? this.map.get(key) : null; + } + + public List> list() { + return new ArrayList>(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> getAll(Object correlationKey) { + Assert.notNull(correlationKey, "'correlationKey' must not be null"); + List> matched = new ArrayList>(); + Collection> values = map.values(); + for (Message message : values) { + if(message.getHeaders().getCorrelationId().equals(correlationKey)){ + matched.add(message); + } + } + return matched; + } + + + public Message post(T payload) { + Message message = MessageBuilder.withPayload(payload).build(); + this.put(message); + return message; + } + +} diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/BufferingMessageHandlerIntegrationTest.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/BufferingMessageHandlerIntegrationTest.java new file mode 100644 index 0000000000..1c967bf6b2 --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/BufferingMessageHandlerIntegrationTest.java @@ -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); + } +} diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/BufferingMessageHandlerTest.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/BufferingMessageHandlerTest.java new file mode 100644 index 0000000000..68f8dbe70d --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/BufferingMessageHandlerTest.java @@ -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> storedMessages = new ArrayList>(); + + 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(); + } +} diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/NewResequencerTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/NewResequencerTests.java new file mode 100644 index 0000000000..237fb60197 --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/NewResequencerTests.java @@ -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(); + } + +} diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/integration/ConcurrentAggregatorIntegrationTests-context.xml b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/integration/ConcurrentAggregatorIntegrationTests-context.xml index 89578eadbc..8aaf94aa0e 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/integration/ConcurrentAggregatorIntegrationTests-context.xml +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/integration/ConcurrentAggregatorIntegrationTests-context.xml @@ -6,8 +6,6 @@ http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd"> - - diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/integration/ConcurrentAggregatorIntegrationTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/integration/ConcurrentAggregatorIntegrationTests.java index f4865d167b..2e9fc8bfa5 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/integration/ConcurrentAggregatorIntegrationTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/integration/ConcurrentAggregatorIntegrationTests.java @@ -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; } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/integration/NewAggregatorEndpointTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/integration/NewAggregatorEndpointTests.java new file mode 100644 index 0000000000..78fdcdd018 --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/integration/NewAggregatorEndpointTests.java @@ -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 reply1 = (Message) replyChannel1.receive(500); + assertNotNull(reply1); + assertThat(reply1.getPayload(), is(105)); + Message reply2 = (Message) 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 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> 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> messagesUpForProcessing, MessageChannel outputChannel, BufferedMessagesCallback processedCallback) { + //noop + } + } +} diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/test/util/TestUtils.java b/org.springframework.integration/src/test/java/org/springframework/integration/test/util/TestUtils.java new file mode 100644 index 0000000000..8c1119cb6b --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/test/util/TestUtils.java @@ -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 getPropertyValue(Object root, String propertyPath, Class 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 messageMatcher) { + return new MessageHandler() { + public void handleMessage(Message message) throws MessageRejectedException, MessageHandlingException, MessageDeliveryException { + assertThat(message, is(messageMatcher)); + } + }; + } +} \ No newline at end of file