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

Merge branch 'work'

Browse Source
This commit is contained in:
David Syer
2010-05-03 08:58:21 +00:00
parent 9850551ddd
commit 8c2c2ca416
35 changed files with 290 additions and 222 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java
View File

@@ -23,6 +23,7 @@ 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.MessageGroup;
import org.springframework.util.Assert;
import java.util.*;

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

@@ -13,7 +13,6 @@
package org.springframework.integration.aggregator;
import java.util.Collection;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
@@ -28,6 +27,8 @@ import org.springframework.integration.core.MessageHeaders;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.handler.AbstractMessageHandler;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.store.MessageGroup;
import org.springframework.integration.store.MessageGroupStore;
import org.springframework.integration.store.MessageStore;
import org.springframework.integration.store.SimpleMessageStore;
import org.springframework.scheduling.TaskScheduler;
@@ -58,7 +59,7 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
public static final long DEFAULT_TIMEOUT = 60000L;
private final MessageStore store;
private final MessageGroupStore store;
private final MessageGroupProcessor outputProcessor;
@@ -76,7 +77,7 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
private final ConcurrentMap<Object, Object> locks = new ConcurrentHashMap<Object, Object>();
public CorrelatingMessageHandler(MessageGroupProcessor processor, MessageStore store,
public CorrelatingMessageHandler(MessageGroupProcessor processor, MessageGroupStore store,
CorrelationStrategy correlationStrategy, ReleaseStrategy releaseStrategy) {
Assert.notNull(store);
Assert.notNull(processor);
@@ -88,7 +89,7 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
this.channelTemplate.setSendTimeout(DEFAULT_SEND_TIMEOUT);
}
public CorrelatingMessageHandler(MessageGroupProcessor processor, MessageStore store) {
public CorrelatingMessageHandler(MessageGroupProcessor processor, MessageGroupStore store) {
this(processor, store, null, null);
}
@@ -160,8 +161,7 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
Object lock = getLock(correlationKey);
synchronized (lock) {
Collection<Message<?>> messages = store.list(correlationKey);
MessageGroup group = new MessageGroup(messages, correlationKey);
MessageGroup group = store.getMessageGroup(correlationKey);
if (group.add(message)) {
@@ -209,12 +209,11 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
Object lock = getLock(correlationKey);
synchronized (lock) {
Collection<Message<?>> all = store.list(correlationKey);
MessageGroup group = new MessageGroup(all, correlationKey);
if (all.size() > 0) {
MessageGroup group = store.getMessageGroup(correlationKey);
if (group.size() > 0) {
// last chance for normal completion
if (releaseStrategy.canRelease(group)) {
outputProcessor.processAndSend(group, channelTemplate, resolveReplyChannel(all.iterator().next(),
outputProcessor.processAndSend(group, channelTemplate, resolveReplyChannel(group.getOne(),
this.outputChannel));
remove(group);
}
@@ -224,15 +223,15 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
logger.info("Processing partially complete messages for key [" + correlationKey + "] to: "
+ outputChannel);
}
outputProcessor.processAndSend(group, channelTemplate, resolveReplyChannel(all.iterator()
.next(), this.outputChannel));
outputProcessor.processAndSend(group, channelTemplate, resolveReplyChannel(group.getOne(),
this.outputChannel));
}
else {
if (logger.isInfoEnabled()) {
logger.info("Discarding partially complete messages for key [" + correlationKey + "] to: "
+ discardChannel);
}
for (Message<?> message : all) {
for (Message<?> message : group.getUnmarked()) {
discardChannel.send(message);
}
}
@@ -250,19 +249,17 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
}
private void mark(MessageGroup group) {
for (Message<?> message : group.getUnmarked()) {
store.mark(group.getCorrelationKey(), message.getHeaders().getId());
}
store.mark(group);
}
private void remove(MessageGroup group) {
Object correlationKey = group.getCorrelationKey();
store.deleteAll(correlationKey);
store.deleteMessageGroup(correlationKey);
locks.remove(correlationKey);
}
private void store(Object correlationKey, Message<?> message) {
store.put(correlationKey, message);
store.addMessageToGroup(correlationKey, message);
}
}

1
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/DefaultAggregatingMessageGroupProcessor.java
View File

@@ -21,6 +21,7 @@ import java.util.Collection;
import java.util.List;
import org.springframework.integration.core.Message;
import org.springframework.integration.store.MessageGroup;
import org.springframework.util.Assert;
/**

2
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageCountReleaseStrategy.java
View File

@@ -13,6 +13,8 @@
package org.springframework.integration.aggregator;
import org.springframework.integration.store.MessageGroup;
/**
* A {@link ReleaseStrategy} that releases only the first <code>n</code> messages, where <code>n</code> is a threshold.
*

1
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageGroupProcessor.java
View File

@@ -2,6 +2,7 @@ package org.springframework.integration.aggregator;
import org.springframework.integration.channel.MessageChannelTemplate;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.store.MessageGroup;
/**
* A processor for <i>correlated</i> groups of messages. When a message group is <i>complete</i> it is passed to the

1
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessor.java
View File

@@ -29,6 +29,7 @@ import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.integration.annotation.Aggregator;
import org.springframework.integration.annotation.Header;
import org.springframework.integration.core.Message;
import org.springframework.integration.store.MessageGroup;
import org.springframework.util.Assert;
import org.springframework.util.ReflectionUtils;

1
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/PassThroughMessageGroupProcessor.java
View File

@@ -3,6 +3,7 @@ package org.springframework.integration.aggregator;
import org.springframework.integration.channel.MessageChannelTemplate;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.store.MessageGroup;
/**
* This implementation of MessageGroupProcessor will forward all messages inside the group to the given output channel.

2
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/ReleaseStrategy.java
View File

@@ -16,6 +16,8 @@
package org.springframework.integration.aggregator;
import org.springframework.integration.store.MessageGroup;
/**

1
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/ReleaseStrategyAdapter.java
View File

@@ -18,6 +18,7 @@ package org.springframework.integration.aggregator;
import java.lang.reflect.Method;
import org.springframework.integration.store.MessageGroup;
import org.springframework.util.Assert;
/**

1
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/Resequencer.java
View File

@@ -25,6 +25,7 @@ import java.util.List;
import org.springframework.integration.channel.MessageChannelTemplate;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.store.MessageGroup;
/**
* This class implements all the strategy interfaces needed for a default

2
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/SequenceSizeReleaseStrategy.java
View File

@@ -16,6 +16,8 @@
package org.springframework.integration.aggregator;
import org.springframework.integration.store.MessageGroup;
/**
* An implementation of {@link ReleaseStrategy} that simply compares the
* current size of the message list to the expected 'sequenceSize'.

43
org.springframework.integration/src/main/java/org/springframework/integration/store/MessageGroup.java Normal file
View File

@@ -0,0 +1,43 @@
package org.springframework.integration.store;
import java.util.Collection;
import org.springframework.integration.core.Message;
public interface MessageGroup {
/**
* Add a message to the internal list. This is needed to avoid hitting the
* underlying store or copying the internal list. Use with care.
*/
boolean add(Message<?> message);
/**
* @return internal message list, modification is allowed, but not
* recommended
*/
Collection<Message<?>> getUnmarked();
/**
* @return internal message list, modification is allowed, but not
* recommended
*/
Collection<Message<?>> getMarked();
/**
* @return the correlation key that links these messages together according
* to a particular CorrelationStrategy
*/
Object getCorrelationKey();
boolean isComplete();
int getSequenceSize();
int size();
void mark();
public Message<?> getOne();
}

66
org.springframework.integration/src/main/java/org/springframework/integration/store/MessageGroupStore.java Normal file
View File

@@ -0,0 +1,66 @@
/*
* 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.Collection;
import org.springframework.integration.core.Message;
/**
* Interface for storage operations on groups of messages linked by a correlation key.
*
* @author Dave Syer
*
* @since 2.0
*
*/
public interface MessageGroupStore {
/**
* Return all Messages currently in the MessageStore that were stored using
* {@link #addMessageToGroup(Object, Collection)} with this correlation id.
*
* @return a group of messages, empty if none exists for this key
*
* @see org.springframework.integration.core.MessageHeaders#getCorrelationId()
*/
MessageGroup getMessageGroup(Object correlationId);
/**
* Store a message with an association to a correlation id. This can be used to group messages together instead of
* storing them just under their id.
*
* @param correlationId the correlation id to store the message under
* @param message a message
*/
void addMessageToGroup(Object correlationId, Message<?> message);
/**
* Persist the mark on all the messages from the group. The group is modified in the process as all its unmarked
* messages become marked.
*
* @param group a MessageGroup with no unmarked messages
*/
void mark(MessageGroup group);
/**
* Delete all the messages from the association with this correlation id.
*
* @param correlationId the correlation id to remove
*/
void deleteMessageGroup(Object correlationId);
}

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

@@ -16,78 +16,40 @@
package org.springframework.integration.store;
import java.util.Collection;
import java.util.UUID;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageHeaders;
/**
* 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.
* Strategy interface for storing and retrieving messages.
*
* @author Mark Fisher
* @author Iwein Fuld
* @author Dave Syer
*
* @since 2.0
*/
public interface MessageStore {
String PROCESSED = MessageHeaders.PREFIX + "processed";
/**
* Return the Message with the given id, or <i>null</i> if no Message with that id exists in the MessageStore.
*/
Message<?> get(UUID id);
Message<?> getMessage(UUID id);
/**
* Put the provided Message into the MessageStore. The store may need to mutate the message internally, and if it
* does then the return value can be different than the input. The id of the return value will be used as an index
* so that the {@link #get(UUID)} and {@link #delete(Object)} behave properly. Since messages are immutable, putting
* so that the {@link #getMessage(UUID)} and {@link #removeMessage(UUID)} behave properly. Since messages are immutable, putting
* the same message more than once is a no-op.
*
* @return the message that was stored
*/
<T> Message<T> put(Message<T> message);
<T> Message<T> addMessage(Message<T> message);
/**
* 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(UUID id);
/**
* Return all Messages currently in the MessageStore that were stored using {@link #put(Object, Message)} or
* {@link #put(Object, Collection)} with this correlation id.
*
* @see org.springframework.integration.core.MessageHeaders#getCorrelationId()
*/
Collection<Message<?>> list(Object correlationId);
/**
* Store a message with an association to a correlation id. This can be used to group messages together instead of
* storing them just under their id.
*
* @param correlationId the correlation id to store the message under
* @param message a message
*/
void put(Object correlationId, Message<?> message);
/**
* Mark a message from the association with this correlation id. If the message was previously added using
* {@link #put(Object, Message)} then it will be removed and re-inserted with the {@value #PROCESSED} flag set in
* the headers.
*
* @param correlationId the correlation id to mark the message under
*/
Message<?> mark(Object correlationId, UUID messageId);
/**
* Delete all the messages from the association with this correlation id. If the messages were stored under their id
* through {@link #put(Message)} they are still accessible via {@link #get(UUID)}.
*
* @param correlationId the correlation id to delete all messages under
*/
void deleteAll(Object correlationId);
Message<?> removeMessage(UUID id);
}

76
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageGroup.java → org.springframework.integration/src/main/java/org/springframework/integration/store/SimpleMessageGroup.java
View File

@@ -11,18 +11,18 @@
* specific language governing permissions and limitations under the License.
*/
package org.springframework.integration.aggregator;
package org.springframework.integration.store;
import java.util.Collection;
import java.util.HashSet;
import org.springframework.integration.core.Message;
import org.springframework.integration.store.MessageStore;
/**
* Represents a mutable group of correlated messages that is bound to a certain {@link MessageStore} and correlation
* key. The group will grow during its lifetime, when messages are <code>add</code>ed to it. <strong>This is not thread
* safe and should not be used for long running aggregations</strong>.
* Represents a mutable group of correlated messages that is bound to a certain
* {@link MessageStore} and correlation key. The group will grow during its
* lifetime, when messages are <code>add</code>ed to it. <strong>This is not
* thread safe and should not be used for long running aggregations</strong>.
*
* @author Iwein Fuld
* @author Oleg Zhurakousky
@@ -30,59 +30,48 @@ import org.springframework.integration.store.MessageStore;
*
* @since 2.0
*/
public class MessageGroup {
public class SimpleMessageGroup implements MessageGroup {
private final Object correlationKey;
private final Collection<Message<?>> marked = new HashSet<Message<?>>();
public final Collection<Message<?>> marked = new HashSet<Message<?>>();
private final Collection<Message<?>> unmarked = new HashSet<Message<?>>();
public final Collection<Message<?>> unmarked = new HashSet<Message<?>>();
public MessageGroup(Object correlationKey) {
public SimpleMessageGroup(Object correlationKey) {
this.correlationKey = correlationKey;
}
public MessageGroup(Collection<? extends Message<?>> originalMessages, Object correlationKey) {
public SimpleMessageGroup(MessageGroup template) {
this.correlationKey = template.getCorrelationKey();
this.marked.addAll(template.getMarked());
this.unmarked.addAll(template.getUnmarked());
}
public SimpleMessageGroup(Collection<? extends Message<?>> originalMessages,
Object correlationKey) {
this(correlationKey);
for (Message<?> message : originalMessages) {
add(message);
}
}
/**
* Add a message to the internal list. This is needed to avoid hitting the underlying store or copying the internal
* list. Use with care.
*/
public boolean add(Message<?> message) {
if (isMember(message)) {
return false;
}
if (message.getHeaders().containsKey(MessageStore.PROCESSED)
&& message.getHeaders().get(MessageStore.PROCESSED, Boolean.class)) {
this.marked.add(message);
} else {
this.unmarked.add(message);
}
this.unmarked.add(message);
return true;
}
/**
* @return internal message list, modification is allowed, but not recommended
*/
public Collection<Message<?>> getUnmarked() {
return unmarked;
}
/**
* @return internal message list, modification is allowed, but not recommended
*/
public Collection<Message<?>> getMarked() {
return marked;
}
/**
* @return the correlation key that links these messages together according to a particular CorrelationStrategy
*/
public Object getCorrelationKey() {
return correlationKey;
}
@@ -103,26 +92,35 @@ public class MessageGroup {
return getOne().getHeaders().getSequenceSize();
}
private Message<?> getOne() {
return unmarked.isEmpty() ? (marked.isEmpty() ? null : marked.iterator().next()) : unmarked.iterator().next();
public void mark() {
marked.addAll(unmarked);
unmarked.clear();
}
public int size() {
return marked.size() + unmarked.size();
}
public Message<?> getOne() {
return unmarked.isEmpty() ? (marked.isEmpty() ? null : marked
.iterator().next()) : unmarked.iterator().next();
}
/**
* This method determines whether messages have been added to this group that supersede the given message based on
* its sequence id. This can be helpful to avoid ending up with sequences larger than their required sequence size
* or sequences that are missing certain sequence numbers.
* This method determines whether messages have been added to this group
* that supersede the given message based on its sequence id. This can be
* helpful to avoid ending up with sequences larger than their required
* sequence size or sequences that are missing certain sequence numbers.
*/
private boolean isMember(Message<?> message) {
if (size() == 0) {
return false;
}
Integer messageSequenceNumber = message.getHeaders().getSequenceNumber();
Integer messageSequenceNumber = message.getHeaders()
.getSequenceNumber();
if (messageSequenceNumber != null && messageSequenceNumber > 0) {
Integer messageSequenceSize = message.getHeaders().getSequenceSize();
Integer messageSequenceSize = message.getHeaders()
.getSequenceSize();
if (!messageSequenceSize.equals(getSequenceSize())
|| containsSequenceNumber(unmarked, messageSequenceNumber)
|| containsSequenceNumber(marked, messageSequenceNumber)) {
@@ -132,9 +130,11 @@ public class MessageGroup {
return false;
}
private boolean containsSequenceNumber(Collection<Message<?>> messages, Integer messageSequenceNumber) {
private boolean containsSequenceNumber(Collection<Message<?>> messages,
Integer messageSequenceNumber) {
for (Message<?> member : messages) {
Integer memberSequenceNumber = member.getHeaders().getSequenceNumber();
Integer memberSequenceNumber = member.getHeaders()
.getSequenceNumber();
if (messageSequenceNumber.equals(memberSequenceNumber)) {
return true;
}

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

@@ -16,43 +16,40 @@
package org.springframework.integration.store;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.Iterator;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessagingException;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.util.UpperBound;
import org.springframework.util.Assert;
/**
* Map-based implementation of {@link MessageStore} that enforces a maximum
* capacity.
* Map-based implementation of {@link MessageStore} and {@link MessageGroupStore}. Enforces a maximum capacity for the
* store.
*
* @author Iwein Fuld
* @author Mark Fisher
* @author Dave Syer
*
* @since 2.0
*/
public class SimpleMessageStore implements MessageStore {
public class SimpleMessageStore implements MessageStore, MessageGroupStore {
private final ConcurrentMap<UUID, Message<?>> idToMessage;
private final ConcurrentMap<Object, Collection<Message<?>>> correlationToMessage;
private final ConcurrentMap<Object, SimpleMessageGroup> correlationToMessageGroup;
private final UpperBound upperBound;
/**
* Creates a SimpleMessageStore with a maximum size limited by the given
* capacity, or unlimited size if the given capacity is less than 1.
* Creates a SimpleMessageStore with a maximum size limited by the given capacity, or unlimited size if the given
* capacity is less than 1.
*/
public SimpleMessageStore(int capacity) {
this.idToMessage = new ConcurrentHashMap<UUID, Message<?>>();
this.correlationToMessage = new ConcurrentHashMap<Object, Collection<Message<?>>>();
this.correlationToMessageGroup = new ConcurrentHashMap<Object, SimpleMessageGroup>();
this.upperBound = new UpperBound(capacity);
}
@@ -64,23 +61,23 @@ public class SimpleMessageStore implements MessageStore {
}
@SuppressWarnings("unchecked")
public <T> Message<T> put(Message<T> message) {
public <T> Message<T> addMessage(Message<T> message) {
if (!upperBound.tryAcquire(0)) {
throw new MessagingException(this.getClass().getSimpleName()
+ " was out of capacity at, try constructing it with a larger capacity.");
}
Object correlationId = message.getHeaders().getCorrelationId();
if (correlationId!=null) {
getMessagesInternal(correlationId).add(message);
if (correlationId != null) {
getMessageGroupInternal(correlationId).add(message);
}
return (Message<T>) this.idToMessage.put(message.getHeaders().getId(), message);
}
public Message<?> get(UUID key) {
public Message<?> getMessage(UUID key) {
return (key != null) ? this.idToMessage.get(key) : null;
}
public Message<?> delete(UUID key) {
public Message<?> removeMessage(UUID key) {
if (key != null) {
upperBound.release();
return this.idToMessage.remove(key);
@@ -93,46 +90,35 @@ public class SimpleMessageStore implements MessageStore {
return this.idToMessage.size();
}
public Collection<Message<?>> list(Object correlationId) {
public MessageGroup getMessageGroup(Object correlationId) {
Assert.notNull(correlationId, "'correlationKey' must not be null");
Collection<Message<?>> collection = correlationToMessage.get(correlationId);
if (collection==null) {
return Collections.emptySet();
MessageGroup collection = correlationToMessageGroup.get(correlationId);
if (collection == null) {
return new SimpleMessageGroup(correlationId);
}
return new HashSet<Message<?>>(collection);
return new SimpleMessageGroup(collection);
}
public void put(Object correlationId, Message<?> message) {
getMessagesInternal(correlationId).add(message);
public void addMessageToGroup(Object correlationId, Message<?> message) {
getMessageGroupInternal(correlationId).add(message);
}
public void mark(MessageGroup group) {
Object correlationId = group.getCorrelationKey();
MessageGroup internal = getMessageGroupInternal(correlationId);
internal.mark();
group.mark();
}
public Message<?> mark(Object correlationId, UUID messageId) {
if (!correlationToMessage.containsKey(correlationId)) {
return null;
}
Collection<Message<?>> messages = getMessagesInternal(correlationId);
Message<?> result = null;
for (Iterator<Message<?>> iterator = messages.iterator(); iterator.hasNext();) {
Message<?> message = (Message<?>) iterator.next();
if (message.getHeaders().getId().equals(messageId)) {
iterator.remove();
result = MessageBuilder.fromMessage(message).setHeader(PROCESSED, true).build();
}
}
messages.add(result);
return result;
public void deleteMessageGroup(Object correlationId) {
correlationToMessageGroup.remove(correlationId);
}
public void deleteAll(Object correlationId) {
correlationToMessage.remove(correlationId);
}
private Collection<Message<?>> getMessagesInternal(Object correlationId) {
if (!correlationToMessage.containsKey(correlationId)) {
correlationToMessage.putIfAbsent(correlationId, new HashSet<Message<?>>());
private MessageGroup getMessageGroupInternal(Object correlationId) {
if (!correlationToMessageGroup.containsKey(correlationId)) {
correlationToMessageGroup.putIfAbsent(correlationId, new SimpleMessageGroup(correlationId));
}
Collection<Message<?>> collection = correlationToMessage.get(correlationId);
return collection;
return correlationToMessageGroup.get(correlationId);
}
}

4
org.springframework.integration/src/main/java/org/springframework/integration/transformer/ClaimCheckTransformer.java
View File

@@ -59,13 +59,13 @@ public class ClaimCheckTransformer extends AbstractTransformer {
private Message<?> retrieveMessage(UUID id) {
Message<?> result = this.messageStore.get(id);
Message<?> result = this.messageStore.getMessage(id);
Assert.notNull(result, "unable to locate Message for claim check ID: " + id);
return result;
}
private void storeMessage(Message<?> message) {
this.messageStore.put(message);
this.messageStore.addMessage(message);
}
}

12
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/AggregatingMessageGroupProcessorHeaderTests.java
View File

@@ -31,6 +31,8 @@ import org.springframework.integration.channel.MessageChannelTemplate;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.core.Message;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.store.MessageGroup;
import org.springframework.integration.store.SimpleMessageGroup;
/**
* @author Mark Fisher
@@ -104,7 +106,7 @@ public class AggregatingMessageGroupProcessorHeaderTests {
headers.put("k2", new Integer(2));
Message<?> message = correlatedMessage(1, 1, 1, headers);
List<Message<?>> messages = Collections.<Message<?>>singletonList(message);
MessageGroup group = new MessageGroup(messages, 1);
MessageGroup group = new SimpleMessageGroup(messages, 1);
processor.processAndSend(group, channelTemplate, outputChannel);
Message<?> result = outputChannel.receive(0);
assertNotNull(result);
@@ -119,7 +121,7 @@ public class AggregatingMessageGroupProcessorHeaderTests {
Message<?> message1 = correlatedMessage(1, 2, 1, headers);
Message<?> message2 = correlatedMessage(1, 2, 2, headers);
List<Message<?>> messages = Arrays.<Message<?>>asList(message1, message2);
MessageGroup group = new MessageGroup(messages, 1);
MessageGroup group = new SimpleMessageGroup(messages, 1);
processor.processAndSend(group, channelTemplate, outputChannel);
Message<?> result = outputChannel.receive(0);
assertNotNull(result);
@@ -137,7 +139,7 @@ public class AggregatingMessageGroupProcessorHeaderTests {
headers2.put("k2", new Integer(123));
Message<?> message2 = correlatedMessage(1, 2, 2, headers2);
List<Message<?>> messages = Arrays.<Message<?>>asList(message1, message2);
MessageGroup group = new MessageGroup(messages, 1);
MessageGroup group = new SimpleMessageGroup(messages, 1);
processor.processAndSend(group, channelTemplate, outputChannel);
Message<?> result = outputChannel.receive(0);
assertNotNull(result);
@@ -167,7 +169,7 @@ public class AggregatingMessageGroupProcessorHeaderTests {
headers3.put("conflictBetween2And3", "valueFor3");
Message<?> message3 = correlatedMessage(1, 3, 3, headers3);
List<Message<?>> messages = Arrays.<Message<?>>asList(message1, message2, message3);
MessageGroup group = new MessageGroup(messages, 1);
MessageGroup group = new SimpleMessageGroup(messages, 1);
processor.processAndSend(group, channelTemplate, outputChannel);
Message<?> result = outputChannel.receive(0);
assertNotNull(result);
@@ -195,7 +197,7 @@ public class AggregatingMessageGroupProcessorHeaderTests {
headers3.put("common", "valueForAll");
Message<?> message3 = correlatedMessage(1, 3, 3, headers3);
List<Message<?>> messages = Arrays.<Message<?>>asList(message1, message2, message3);
MessageGroup group = new MessageGroup(messages, 1);
MessageGroup group = new SimpleMessageGroup(messages, 1);
processor.processAndSend(group, channelTemplate, outputChannel);
Message<?> result = outputChannel.receive(0);
assertNotNull(result);

1
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java
View File

@@ -35,6 +35,7 @@ import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.core.MessageHeaders;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.message.MessageHandlingException;
import org.springframework.integration.store.MessageGroup;
import org.springframework.integration.store.SimpleMessageStore;
/**

1
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/ConcurrentAggregatorTests.java
View File

@@ -38,6 +38,7 @@ 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.store.MessageGroup;
import org.springframework.integration.store.SimpleMessageStore;
/**

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

@@ -25,12 +25,12 @@ import org.junit.Test;
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.MessageGroupStore;
import org.springframework.integration.store.SimpleMessageStore;
public class CorrelatingMessageHandlerIntegrationTest {
private MessageStore store = new SimpleMessageStore(100);
private MessageGroupStore store = new SimpleMessageStore(100);
private MessageChannel outputChannel = mock(MessageChannel.class);

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

@@ -40,6 +40,7 @@ import org.springframework.integration.channel.MessageChannelTemplate;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.store.SimpleMessageGroup;
import org.springframework.integration.store.SimpleMessageStore;
import org.springframework.test.util.ReflectionTestUtils;
@@ -68,7 +69,7 @@ public class CorrelatingMessageHandlerTests {
handler = new CorrelatingMessageHandler(processor, new SimpleMessageStore(), correlationStrategy,
ReleaseStrategy);
handler.setOutputChannel(outputChannel);
doAnswer(new DoesNothing()).when(processor).processAndSend(isA(MessageGroup.class),
doAnswer(new DoesNothing()).when(processor).processAndSend(isA(SimpleMessageGroup.class),
isA(MessageChannelTemplate.class), eq(outputChannel));
}
@@ -91,7 +92,7 @@ public class CorrelatingMessageHandlerTests {
verify(correlationStrategy).getCorrelationKey(message1);
verify(correlationStrategy).getCorrelationKey(message2);
verify(processor).processAndSend(isA(MessageGroup.class), isA(MessageChannelTemplate.class), eq(outputChannel));
verify(processor).processAndSend(isA(SimpleMessageGroup.class), isA(MessageChannelTemplate.class), eq(outputChannel));
}
private void verifyLocks(CorrelatingMessageHandler handler, int lockCount) {

1
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessorTests.java
View File

@@ -30,6 +30,7 @@ import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.store.MessageGroup;
@RunWith(MockitoJUnitRunner.class)
public class MethodInvokingMessageGroupProcessorTests {

4
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/ReleaseStrategyAdapterTests.java
View File

@@ -25,6 +25,8 @@ import org.junit.Before;
import org.junit.Test;
import org.springframework.integration.core.Message;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.store.MessageGroup;
import org.springframework.integration.store.SimpleMessageGroup;
/**
* @author Marius Bogoevici
@@ -155,7 +157,7 @@ public class ReleaseStrategyAdapterTests {
if (size > 2) {
messages.add(new GenericMessage<String>("789"));
}
return new MessageGroup(messages, "ABC");
return new SimpleMessageGroup(messages, "ABC");
}
@SuppressWarnings("unused")

8
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java
View File

@@ -19,7 +19,6 @@ package org.springframework.integration.aggregator;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import java.util.ArrayList;
import java.util.Arrays;
@@ -31,12 +30,13 @@ 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.store.MessageStore;
import org.springframework.integration.store.MessageGroupStore;
import org.springframework.integration.store.SimpleMessageStore;
/**
* @author Marius Bogoevici
* @author Alex Peters
* @author Dave Syer
*/
public class ResequencerTests {
@@ -44,7 +44,7 @@ public class ResequencerTests {
private Resequencer processor = new Resequencer();
private MessageStore store = new SimpleMessageStore();
private MessageGroupStore store = new SimpleMessageStore();
@Before
public void configureResequencer() {
@@ -230,7 +230,7 @@ public class ResequencerTests {
String correlationId = "ABC";
Message<?> message1 = createMessage("123", correlationId, 1, 1, replyChannel);
resequencer.handleMessage(message1);
assertTrue(store.list(correlationId).isEmpty());
assertEquals(0, store.getMessageGroup(correlationId).size());
}
private static Message<?> createMessage(String payload, Object correlationId, int sequenceSize, int sequenceNumber,

8
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/SequenceSizeReleaseStrategyTests.java
View File

@@ -22,6 +22,8 @@ import static org.junit.Assert.assertTrue;
import org.junit.Test;
import org.springframework.integration.core.Message;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.store.MessageGroup;
import org.springframework.integration.store.SimpleMessageGroup;
/**
* @author Mark Fisher
@@ -32,7 +34,7 @@ public class SequenceSizeReleaseStrategyTests {
public void testIncompleteList() {
Message<String> message = MessageBuilder.withPayload("test1")
.setSequenceSize(2).build();
MessageGroup messages = new MessageGroup("FOO");
MessageGroup messages = new SimpleMessageGroup("FOO");
messages.add(message);
SequenceSizeReleaseStrategy ReleaseStrategy = new SequenceSizeReleaseStrategy();
assertFalse(ReleaseStrategy.canRelease(messages));
@@ -44,7 +46,7 @@ public class SequenceSizeReleaseStrategyTests {
.setSequenceSize(2).build();
Message<String> message2 = MessageBuilder.withPayload("test2")
.setSequenceSize(2).build();
MessageGroup messages = new MessageGroup("FOO");
MessageGroup messages = new SimpleMessageGroup("FOO");
messages.add(message1);
messages.add(message2);
SequenceSizeReleaseStrategy ReleaseStrategy = new SequenceSizeReleaseStrategy();
@@ -54,7 +56,7 @@ public class SequenceSizeReleaseStrategyTests {
@Test
public void testEmptyList() {
SequenceSizeReleaseStrategy ReleaseStrategy = new SequenceSizeReleaseStrategy();
assertTrue(ReleaseStrategy.canRelease(new MessageGroup("FOO")));
assertTrue(ReleaseStrategy.canRelease(new SimpleMessageGroup("FOO")));
}
}

2
org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorWithCorrelationStrategyTests.java
View File

@@ -21,7 +21,6 @@ import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.integration.aggregator.MessageGroup;
import org.springframework.integration.aggregator.ReleaseStrategy;
import org.springframework.integration.aggregator.CorrelationStrategy;
import org.springframework.integration.annotation.Aggregator;
@@ -29,6 +28,7 @@ import org.springframework.integration.channel.PollableChannel;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.store.MessageGroup;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

2
org.springframework.integration/src/test/java/org/springframework/integration/config/TestReleaseStrategy.java
View File

@@ -17,8 +17,8 @@
package org.springframework.integration.config;
import org.springframework.integration.aggregator.MessageGroup;
import org.springframework.integration.aggregator.ReleaseStrategy;
import org.springframework.integration.store.MessageGroup;
/**
* @author Marius Bogoevici

7
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/MessageGroupTests.java → org.springframework.integration/src/test/java/org/springframework/integration/store/SimpleMessageGroupTests.java
View File

@@ -1,4 +1,4 @@
package org.springframework.integration.aggregator;
package org.springframework.integration.store;
import static org.hamcrest.CoreMatchers.is;
import static org.junit.Assert.assertThat;
@@ -9,13 +9,14 @@ import org.junit.Before;
import org.junit.Test;
import org.springframework.integration.core.Message;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.store.SimpleMessageGroup;
/**
* @author Iwein Fuld
* @author Oleg Zhurakousky
* @author Dave Syer
*/
public class MessageGroupTests {
public class SimpleMessageGroupTests {
private Object key = new Object();
@@ -23,7 +24,7 @@ public class MessageGroupTests {
@Before
public void buildMessageGroup() {
group = new MessageGroup(Collections.<Message<?>> emptyList(), key);
group = new SimpleMessageGroup(Collections.<Message<?>> emptyList(), key);
}
@Test

25
org.springframework.integration/src/test/java/org/springframework/integration/store/SimpleMessageStoreTests.java
View File

@@ -18,6 +18,7 @@ package org.springframework.integration.store;
import static org.hamcrest.CoreMatchers.is;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotSame;
import static org.junit.Assert.assertThat;
import org.junit.Test;
@@ -36,8 +37,8 @@ public class SimpleMessageStoreTests {
public void shouldRetainMessage() {
SimpleMessageStore store = new SimpleMessageStore();
Message<String> testMessage1 = MessageBuilder.withPayload("foo").build();
store.put(testMessage1);
assertThat((Message<String>) store.get(testMessage1.getHeaders().getId()), is(testMessage1));
store.addMessage(testMessage1);
assertThat((Message<String>) store.getMessage(testMessage1.getHeaders().getId()), is(testMessage1));
}
@Test(expected = MessagingException.class)
@@ -45,8 +46,8 @@ public class SimpleMessageStoreTests {
SimpleMessageStore store = new SimpleMessageStore(1);
Message<String> testMessage1 = MessageBuilder.withPayload("foo").build();
Message<String> testMessage2 = MessageBuilder.withPayload("bar").build();
store.put(testMessage1);
store.put(testMessage2);
store.addMessage(testMessage1);
store.addMessage(testMessage2);
}
@Test
@@ -54,16 +55,24 @@ public class SimpleMessageStoreTests {
SimpleMessageStore store = new SimpleMessageStore(2);
Message<String> testMessage1 = MessageBuilder.withPayload("foo").build();
Message<String> testMessage2 = MessageBuilder.withPayload("bar").build();
store.put(testMessage1);
store.put(testMessage2);
store.addMessage(testMessage1);
store.addMessage(testMessage2);
}
@Test
public void shouldListByCorrelation() throws Exception {
SimpleMessageStore store = new SimpleMessageStore();
Message<String> testMessage1 = MessageBuilder.withPayload("foo").build();
store.put("bar", testMessage1);
assertEquals(1, store.list("bar").size());
store.addMessageToGroup("bar", testMessage1);
assertEquals(1, store.getMessageGroup("bar").size());
}
@Test
public void shouldCopyMessageGroup() throws Exception {
SimpleMessageStore store = new SimpleMessageStore();
Message<String> testMessage1 = MessageBuilder.withPayload("foo").build();
store.addMessageToGroup("bar", testMessage1);
assertNotSame(store.getMessageGroup("bar"), store.getMessageGroup("bar"));
}
}

2
org.springframework.integration/src/test/java/org/springframework/integration/transformer/ClaimCheckTransformerTests.java
View File

@@ -47,7 +47,7 @@ public class ClaimCheckTransformerTests {
MessageStore store = new SimpleMessageStore(10);
Message<?> message = MessageBuilder.withPayload("test").build();
UUID storedId = message.getHeaders().getId();
store.put(message);
store.addMessage(message);
ClaimCheckTransformer transformer = new ClaimCheckTransformer(store);
Message<?> input = MessageBuilder.withPayload(storedId).build();
Message<?> output = transformer.transform(input);