diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java index 1378b1b4c2..558955a732 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java @@ -13,8 +13,21 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.aggregator; +import java.util.List; +import java.util.Queue; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.DelayQueue; +import java.util.concurrent.Delayed; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.locks.ReentrantLock; + import org.springframework.context.Lifecycle; import org.springframework.integration.channel.ChannelResolver; import org.springframework.integration.channel.NullChannel; @@ -27,15 +40,10 @@ 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.*; -import java.util.concurrent.locks.ReentrantLock; - /** * MessageHandler that holds a buffer of messages in a MessageStore. This class takes care of * correlated groups of messages that can be completed in batches. It is useful for aggregating, - * resequencing, or custom buffering concerns. + * resequencing, or custom implementations requiring correlation. * * @author Iwein Fuld */ @@ -111,7 +119,7 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements Object correlationKey = correlationStrategy.getCorrelationKey(message); try { if (tracker.aquireLockFor(correlationKey)) { - List> group = store.getAll(correlationKey); + List> group = store.list(correlationKey); if (noSupersedingMessage(message, group)) { store(message, correlationKey); group.add(message); @@ -213,7 +221,7 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements } protected final void forceComplete(Object key) { - List> all = store.getAll(key); + List> all = store.list(key); if (all.size() > 0) { //last chance for normal completion MessageChannel outputChannel = resolveReplyChannel(all.get(0), this.outputChannel, this.channelResolver); 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 index 7eeb5bcf1d..efe5140270 100644 --- 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2008 the original author or authors. + * 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. @@ -23,23 +23,46 @@ 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. + * helpful when mapping to a RESTful API. * * @author Mark Fisher * @author Iwein Fuld + * @since 2.0 */ public interface MessageStore { - Message get(Object key); + /** + * Return the Message with the given id, or null if no + * Message with that id exists in the MessageStore. + */ + Message get(Object id); + /** + * Put the provided Message into the MessageStore. Its id will + * be used as an index so that the {@link #get(Object)} and + * {@link #delete(Object)} behave properly. If available, its + * correlationId header will also be stored so that the + * {@link #list(Object)} method behaves properly. + */ Message put(Message message); - Message post(T payload); - - Message delete(Object key); + /** + * Remove the Message with the given id from the MessageStore, + * if present, and return it. If no Message with that id is + * present in the store, this will return null. + */ + Message delete(Object id); + /** + * Return all Messages currently in the MessageStore. + */ List> list(); - List> getAll(Object correlationKey); + /** + * Return all Messages currently in the MessageStore that + * contain the provided correlationId header value. + * @see org.springframework.integration.core.MessageHeaders#getCorrelationId() + */ + List> list(Object correlationId); } 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 index 75c18cfd0d..deccdcfd09 100644 --- 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2008 the original author or authors. + * 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. @@ -23,7 +23,6 @@ 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; /** @@ -31,6 +30,7 @@ import org.springframework.util.Assert; * * @author Iwein Fuld * @author Mark Fisher + * @since 2.0 */ public class SimpleMessageStore implements MessageStore { @@ -43,7 +43,7 @@ public class SimpleMessageStore implements MessageStore { @SuppressWarnings("unchecked") - public Message put( Message message) { + public Message put(Message message) { return (Message) this.map.put(message.getHeaders().getId(), message); } @@ -64,7 +64,7 @@ public class SimpleMessageStore implements MessageStore { } - public List> getAll(Object correlationKey) { + public List> list(Object correlationKey) { Assert.notNull(correlationKey, "'correlationKey' must not be null"); List> matched = new ArrayList>(); Collection> values = map.values(); @@ -76,11 +76,4 @@ public class SimpleMessageStore implements MessageStore { 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/BufferingMessageHandlerTest.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/BufferingMessageHandlerTest.java deleted file mode 100644 index 13726d995a..0000000000 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/BufferingMessageHandlerTest.java +++ /dev/null @@ -1,92 +0,0 @@ -/* - * 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 org.mockito.runners.MockitoJUnitRunner; -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; - -import static org.mockito.Mockito.*; - -/** - * @author Iwein Fuld - */ -@RunWith(MockitoJUnitRunner.class) -public class BufferingMessageHandlerTest { - - private CorrelatingMessageHandler buffer; - @Mock - private MessageStore store; - @Mock - private CorrelationStrategy correlationStrategy; - @Mock - private CompletionStrategy completionStrategy; - @Mock - private MessageGroupProcessor processor; - @Mock - private MessageChannel outputChannel; - - @Before - public void initializeSubject() { - buffer = new CorrelatingMessageHandler(store, correlationStrategy, - completionStrategy, processor); - buffer.setOutputChannel(outputChannel); - } - - @Test - public void bufferCompletesNormally() throws Exception { - String correlationKey = "key"; - Message message1 = testMessage(1, 1); - Message message2 = testMessage(2, 2); - List> storedMessages = new ArrayList>(); - when(store.getAll(correlationKey)).thenReturn(storedMessages); - - when(correlationStrategy.getCorrelationKey(isA(Message.class))) - .thenReturn(correlationKey); - when(completionStrategy.isComplete(storedMessages)).thenReturn(false); - - buffer.handleMessageInternal(message1); - storedMessages.add(message1); - - when(completionStrategy.isComplete(storedMessages)).thenReturn(true); - buffer.handleMessageInternal(message2); - storedMessages.add(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, int sequenceNumber) { - return MessageBuilder.withPayload("test").setHeader(MessageHeaders.ID, - id).setSequenceNumber(sequenceNumber).build(); - } -} diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java new file mode 100644 index 0000000000..ebc5ba67f7 --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java @@ -0,0 +1,109 @@ +/* + * 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 static org.mockito.Matchers.eq; +import static org.mockito.Matchers.isA; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.util.ArrayList; +import java.util.List; + +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mock; +import org.mockito.runners.MockitoJUnitRunner; + +import org.springframework.integration.core.Message; +import org.springframework.integration.core.MessageChannel; +import org.springframework.integration.core.MessageHeaders; +import org.springframework.integration.message.MessageBuilder; +import org.springframework.integration.store.MessageStore; + +/** + * @author Iwein Fuld + */ +@RunWith(MockitoJUnitRunner.class) +public class CorrelatingMessageHandlerTests { + + private CorrelatingMessageHandler handler; + + @Mock + private MessageStore store; + + @Mock + private CorrelationStrategy correlationStrategy; + + @Mock + private CompletionStrategy completionStrategy; + + @Mock + private MessageGroupProcessor processor; + + @Mock + private MessageChannel outputChannel; + + @Before + public void initializeSubject() { + handler = new CorrelatingMessageHandler( + store, correlationStrategy, completionStrategy, processor); + handler.setOutputChannel(outputChannel); + } + + @Test + public void bufferCompletesNormally() throws Exception { + String correlationKey = "key"; + Message message1 = testMessage(1, 1); + Message message2 = testMessage(2, 2); + List> storedMessages = new ArrayList>(); + + when(store.list(correlationKey)).thenReturn(storedMessages); + + when(correlationStrategy.getCorrelationKey(isA(Message.class))) + .thenReturn(correlationKey); + + when(completionStrategy.isComplete(storedMessages)).thenReturn(false); + + handler.handleMessage(message1); + storedMessages.add(message1); + + when(completionStrategy.isComplete(storedMessages)).thenReturn(true); + handler.handleMessage(message2); + storedMessages.add(message2); + + verify(store).put(message1); + verify(store).put(message2); + verify(store, times(2)).list(correlationKey); + verify(correlationStrategy).getCorrelationKey(message1); + verify(correlationStrategy).getCorrelationKey(message2); + verify(completionStrategy, times(2)).isComplete(storedMessages); + verify(processor).processAndSend(eq(correlationKey), + eq(storedMessages), eq(outputChannel), + isA(BufferedMessagesCallback.class)); + } + + + private Message testMessage(int id, int sequenceNumber) { + return MessageBuilder.withPayload("test") + .setHeader(MessageHeaders.ID, id) + .setSequenceNumber(sequenceNumber).build(); + } + +}