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

Renamed MessageStore's getAll(correlationKey) method to 'list(..)'

Browse Source
This commit is contained in:
Mark Fisher
2009-12-02 18:25:38 +00:00
parent 6a0e72bd6b
commit f7bee0902b
5 changed files with 159 additions and 118 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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<Message<?>> group = store.getAll(correlationKey);
List<Message<?>> 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<Message<?>> all = store.getAll(key);
List<Message<?>> all = store.list(key);
if (all.size() > 0) {
//last chance for normal completion
MessageChannel outputChannel = resolveReplyChannel(all.get(0), this.outputChannel, this.channelResolver);

37
org.springframework.integration/src/main/java/org/springframework/integration/store/MessageStore.java
View File

@@ -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 <i>null</i> 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.
*/
<T> Message<T> put(Message<T> message);
<T> Message<T> 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 <i>null</i>.
*/
Message<?> delete(Object id);
/**
* Return all Messages currently in the MessageStore.
*/
List<Message<?>> list();
List<Message<?>> 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<Message<?>> list(Object correlationId);
}

15
org.springframework.integration/src/main/java/org/springframework/integration/store/SimpleMessageStore.java
View File

@@ -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 <T> Message<T> put( Message<T> message) {
public <T> Message<T> put(Message<T> message) {
return (Message<T>) this.map.put(message.getHeaders().getId(), message);
}
@@ -64,7 +64,7 @@ public class SimpleMessageStore implements MessageStore {
}
public List<Message<?>> getAll(Object correlationKey) {
public List<Message<?>> list(Object correlationKey) {
Assert.notNull(correlationKey, "'correlationKey' must not be null");
List<Message<?>> matched = new ArrayList<Message<?>>();
Collection<Message<?>> values = map.values();
@@ -76,11 +76,4 @@ public class SimpleMessageStore implements MessageStore {
return matched;
}
public <T> Message<T> post(T payload) {
Message<T> message = MessageBuilder.withPayload(payload).build();
this.put(message);
return message;
}
}

92
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/BufferingMessageHandlerTest.java
View File

@@ -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<Message<?>> storedMessages = new ArrayList<Message<?>>();
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();
}
}

109
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java Normal file
View File

@@ -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<Message<?>> storedMessages = new ArrayList<Message<?>>();
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();
}
}