INT-330: Added MethodInvokingMessagesProcessor
This commit is contained in:
Iwein Fuld
@@ -16,7 +16,6 @@
|
||||
package org.springframework.integration.aggregator;
|
||||
|
||||
import org.springframework.context.Lifecycle;
|
||||
import org.springframework.integration.channel.ChannelResolutionException;
|
||||
import org.springframework.integration.channel.ChannelResolver;
|
||||
import org.springframework.integration.channel.NullChannel;
|
||||
import org.springframework.integration.core.Message;
|
||||
@@ -34,7 +33,9 @@ import java.util.concurrent.*;
|
||||
import java.util.concurrent.locks.ReentrantLock;
|
||||
|
||||
/**
|
||||
* MessageHandler that holds a buffer of messages in a MessageStore
|
||||
* 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.
|
||||
*
|
||||
* @author Iwein Fuld
|
||||
*/
|
||||
@@ -49,7 +50,7 @@ public class BufferingMessageHandler extends AbstractMessageHandler implements L
|
||||
private volatile MessageChannel discardChannel = new NullChannel();
|
||||
private TaskScheduler taskScheduler;
|
||||
private Object lifecycleMonitor = new Object();
|
||||
private ScheduledFuture reaperFutureTask;
|
||||
private ScheduledFuture<?> reaperFutureTask;
|
||||
private volatile long reaperInterval = 1000l;
|
||||
private final BlockingQueue<DelayedKey> keysInBuffer = new DelayQueue<DelayedKey>();
|
||||
private volatile long timeout = 60000l;
|
||||
|
||||
@@ -16,101 +16,102 @@
|
||||
|
||||
package org.springframework.integration.aggregator;
|
||||
|
||||
import java.lang.reflect.InvocationTargetException;
|
||||
import java.lang.reflect.Method;
|
||||
import java.lang.reflect.ParameterizedType;
|
||||
import java.lang.reflect.Type;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
|
||||
import org.springframework.integration.core.Message;
|
||||
import org.springframework.integration.core.MessagingException;
|
||||
import org.springframework.integration.util.DefaultMethodInvoker;
|
||||
import org.springframework.util.Assert;
|
||||
import org.springframework.util.ReflectionUtils;
|
||||
|
||||
import java.lang.reflect.InvocationTargetException;
|
||||
import java.lang.reflect.Method;
|
||||
import java.lang.reflect.ParameterizedType;
|
||||
import java.lang.reflect.Type;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collection;
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* Base class for implementing adapters for methods which take as an argument a
|
||||
* list of {@link Message Message} instances or payloads.
|
||||
*
|
||||
*
|
||||
* @author Marius Bogoevici
|
||||
*/
|
||||
public class MessageListMethodAdapter {
|
||||
|
||||
private final DefaultMethodInvoker invoker;
|
||||
private final DefaultMethodInvoker invoker;
|
||||
|
||||
protected final Method method;
|
||||
protected final Method method;
|
||||
|
||||
|
||||
public MessageListMethodAdapter(Object object, String methodName) {
|
||||
Assert.notNull(object, "'object' must not be null");
|
||||
Assert.notNull(methodName, "'methodName' must not be null");
|
||||
this.method = ReflectionUtils.findMethod(object.getClass(), methodName, new Class<?>[] { List.class });
|
||||
Assert.notNull(this.method, "Method '" + methodName +
|
||||
"(List<?> args)' not found on '" + object.getClass().getName() + "'.");
|
||||
this.invoker = new DefaultMethodInvoker(object, this.method);
|
||||
}
|
||||
public MessageListMethodAdapter(Object object, String methodName) {
|
||||
Assert.notNull(object, "'object' must not be null");
|
||||
Assert.notNull(methodName, "'methodName' must not be null");
|
||||
this.method = ReflectionUtils.findMethod(object.getClass(), methodName, new Class<?>[]{List.class});
|
||||
Assert.notNull(this.method, "Method '" + methodName +
|
||||
"(List<?> args)' not found on '" + object.getClass().getName() + "'.");
|
||||
this.invoker = new DefaultMethodInvoker(object, this.method);
|
||||
}
|
||||
|
||||
public MessageListMethodAdapter(Object object, Method method) {
|
||||
Assert.notNull(object, "'object' must not be null");
|
||||
Assert.notNull(method, "'method' must not be null");
|
||||
Assert.isTrue(method.getParameterTypes().length == 1
|
||||
&& method.getParameterTypes()[0].equals(List.class),
|
||||
"Method must accept exactly one parameter, and it must be a List.");
|
||||
this.method = method;
|
||||
this.invoker = new DefaultMethodInvoker(object, this.method);
|
||||
}
|
||||
public MessageListMethodAdapter(Object object, Method method) {
|
||||
Assert.notNull(object, "'object' must not be null");
|
||||
Assert.notNull(method, "'method' must not be null");
|
||||
Assert.isTrue(method.getParameterTypes().length == 1
|
||||
&& method.getParameterTypes()[0].equals(List.class),
|
||||
"Method " + method + " does not accept exactly one parameter, of type List.");
|
||||
this.method = method;
|
||||
this.invoker = new DefaultMethodInvoker(object, this.method);
|
||||
}
|
||||
|
||||
|
||||
protected Method getMethod() {
|
||||
return method;
|
||||
}
|
||||
protected Method getMethod() {
|
||||
return method;
|
||||
}
|
||||
|
||||
private static boolean isActualTypeParameterizedMessage(Method method) {
|
||||
return (getCollectionActualType(method) instanceof ParameterizedType)
|
||||
&& Message.class.isAssignableFrom((Class<?>) ((ParameterizedType) getCollectionActualType(method)).getRawType());
|
||||
}
|
||||
private static boolean isActualTypeParameterizedMessage(Method method) {
|
||||
return (getCollectionActualType(method) instanceof ParameterizedType)
|
||||
&& Message.class.isAssignableFrom((Class<?>) ((ParameterizedType) getCollectionActualType(method)).getRawType());
|
||||
}
|
||||
|
||||
protected final Object executeMethod(List<Message<?>> messages) {
|
||||
try {
|
||||
if (isMethodParameterParameterized(this.method) && isHavingActualTypeArguments(this.method)
|
||||
&& (isActualTypeRawMessage(this.method) || isActualTypeParameterizedMessage(this.method))) {
|
||||
return this.invoker.invokeMethod(messages);
|
||||
}
|
||||
return this.invoker.invokeMethod(extractPayloadsFromMessages(messages));
|
||||
}
|
||||
catch (InvocationTargetException e) {
|
||||
throw new MessagingException(
|
||||
"Method '" + this.method + "' threw an Exception.", e.getTargetException());
|
||||
}
|
||||
catch (Exception e) {
|
||||
throw new MessagingException("Failed to invoke method '" + this.method + "'.");
|
||||
}
|
||||
}
|
||||
protected final Object executeMethod(Collection<Message<?>> messages) {
|
||||
try {
|
||||
if (isMethodParameterParameterized(this.method) && isHavingActualTypeArguments(this.method)
|
||||
&& (isActualTypeRawMessage(this.method) || isActualTypeParameterizedMessage(this.method))) {
|
||||
return this.invoker.invokeMethod(messages);
|
||||
}
|
||||
return this.invoker.invokeMethod(extractPayloadsFromMessages(messages));
|
||||
}
|
||||
catch (InvocationTargetException e) {
|
||||
throw new MessagingException(
|
||||
"Method '" + this.method + "' threw an Exception.", e.getTargetException());
|
||||
}
|
||||
catch (Exception e) {
|
||||
throw new MessagingException("Failed to invoke method '" + this.method + "'.");
|
||||
}
|
||||
}
|
||||
|
||||
private List<?> extractPayloadsFromMessages(List<Message<?>> messages) {
|
||||
List<Object> payloadList = new ArrayList<Object>();
|
||||
for (Message<?> message : messages) {
|
||||
payloadList.add(message.getPayload());
|
||||
}
|
||||
return payloadList;
|
||||
}
|
||||
private List<?> extractPayloadsFromMessages(Collection<Message<?>> messages) {
|
||||
List<Object> payloadList = new ArrayList<Object>();
|
||||
for (Message<?> message : messages) {
|
||||
payloadList.add(message.getPayload());
|
||||
}
|
||||
return payloadList;
|
||||
}
|
||||
|
||||
private static boolean isActualTypeRawMessage(Method method) {
|
||||
return getCollectionActualType(method).equals(Message.class);
|
||||
}
|
||||
private static boolean isActualTypeRawMessage(Method method) {
|
||||
return getCollectionActualType(method).equals(Message.class);
|
||||
}
|
||||
|
||||
private static Type getCollectionActualType(Method method) {
|
||||
return ((ParameterizedType) method.getGenericParameterTypes()[0]).getActualTypeArguments()[0];
|
||||
}
|
||||
private static Type getCollectionActualType(Method method) {
|
||||
return ((ParameterizedType) method.getGenericParameterTypes()[0]).getActualTypeArguments()[0];
|
||||
}
|
||||
|
||||
private static boolean isHavingActualTypeArguments(Method method) {
|
||||
return ((ParameterizedType) method.getGenericParameterTypes()[0]).getActualTypeArguments().length == 1;
|
||||
}
|
||||
private static boolean isHavingActualTypeArguments(Method method) {
|
||||
return ((ParameterizedType) method.getGenericParameterTypes()[0]).getActualTypeArguments().length == 1;
|
||||
}
|
||||
|
||||
private static boolean isMethodParameterParameterized(Method method) {
|
||||
return method.getGenericParameterTypes().length == 1
|
||||
&& method.getGenericParameterTypes()[0] instanceof ParameterizedType;
|
||||
}
|
||||
private static boolean isMethodParameterParameterized(Method method) {
|
||||
return method.getGenericParameterTypes().length == 1
|
||||
&& method.getGenericParameterTypes()[0] instanceof ParameterizedType;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -0,0 +1,48 @@
|
||||
package org.springframework.integration.aggregator;
|
||||
|
||||
import org.springframework.beans.factory.FactoryBean;
|
||||
import org.springframework.beans.factory.InitializingBean;
|
||||
import org.springframework.integration.core.MessageHeaders;
|
||||
import org.springframework.integration.handler.MethodInvokingMessageProcessor;
|
||||
import org.springframework.integration.store.MessageStore;
|
||||
import org.springframework.integration.store.SimpleMessageStore;
|
||||
|
||||
public class MethodInvokingAggregatorFactoryBean implements
|
||||
FactoryBean<BufferingMessageHandler> , InitializingBean{
|
||||
|
||||
private static final int DEFAULT_CAPACITY = Integer.MAX_VALUE;
|
||||
|
||||
private MessageStore store = new SimpleMessageStore(DEFAULT_CAPACITY);
|
||||
|
||||
private CorrelationStrategy correlationStrategy = new HeaderAttributeCorrelationStrategy(
|
||||
MessageHeaders.CORRELATION_ID);
|
||||
|
||||
private CompletionStrategy completionStrategy = new SequenceSizeCompletionStrategy();
|
||||
|
||||
private MessagesProcessor processor;
|
||||
|
||||
private Object target;
|
||||
|
||||
public void setTarget(Object target) {
|
||||
this.target = target;
|
||||
}
|
||||
|
||||
public void afterPropertiesSet() throws Exception {
|
||||
// build correllation strategy
|
||||
// build completion strategy
|
||||
// build processor
|
||||
}
|
||||
|
||||
public BufferingMessageHandler getObject() throws Exception {
|
||||
return new BufferingMessageHandler(store, correlationStrategy,
|
||||
completionStrategy, processor);
|
||||
}
|
||||
|
||||
public Class<? extends BufferingMessageHandler> getObjectType() {
|
||||
return BufferingMessageHandler.class;
|
||||
}
|
||||
|
||||
public boolean isSingleton() {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,148 @@
|
||||
package org.springframework.integration.aggregator;
|
||||
|
||||
import org.springframework.integration.annotation.Aggregator;
|
||||
import org.springframework.integration.annotation.Header;
|
||||
import org.springframework.integration.core.Message;
|
||||
import org.springframework.integration.core.MessageChannel;
|
||||
import org.springframework.integration.message.MessageBuilder;
|
||||
import org.springframework.util.Assert;
|
||||
|
||||
import java.lang.annotation.Annotation;
|
||||
import java.lang.reflect.Method;
|
||||
import java.util.*;
|
||||
|
||||
public class MethodInvokingMessagesProcessor implements MessagesProcessor {
|
||||
|
||||
private final Object target;
|
||||
private final Method method;
|
||||
|
||||
private final MessageListMethodAdapter adapter;
|
||||
|
||||
public MethodInvokingMessagesProcessor(Object target) {
|
||||
this.target = target;
|
||||
this.method = selectMethodFrom(target);
|
||||
this.adapter = new MessageListMethodAdapter(target, method);
|
||||
}
|
||||
|
||||
private Method selectMethodFrom(Object target) {
|
||||
Method[] methods = target.getClass().getMethods();
|
||||
Set<Method> candidates = new HashSet<Method>(Arrays.asList(methods));
|
||||
|
||||
removeObjectMethodsFrom(candidates);
|
||||
removeVoidMethodsFrom(candidates);
|
||||
removeListIncompatibleMethodsFrom(candidates);
|
||||
Set<Method> unanotatedCandidates = new HashSet<Method>();
|
||||
if (candidates.size() > 1) {
|
||||
unanotatedCandidates.addAll(removeUnanotatedFrom(candidates));
|
||||
}
|
||||
//if no methods are annotated we need to look in more detail in the unannotated methods
|
||||
if (candidates.size() < 1) {
|
||||
candidates = unanotatedCandidates;
|
||||
removeUnfittingFrom(candidates);
|
||||
}
|
||||
|
||||
Assert.state(candidates.size() == 1,
|
||||
"Method selection failed, there should be exactly one candidate, found ["
|
||||
+ candidates + "]");
|
||||
return candidates.iterator().next();
|
||||
}
|
||||
|
||||
private void removeListIncompatibleMethodsFrom(Set<Method> candidates) {
|
||||
removeMethodsMatchingSelector(candidates, new MethodSelector() {
|
||||
public boolean select(Method method) {
|
||||
int found = 0;
|
||||
for (Class<?> parameterClass : method.getParameterTypes()) {
|
||||
if (parameterClass.isAssignableFrom(List.class)) {
|
||||
found++;
|
||||
}
|
||||
}
|
||||
return found != 1;
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
private void removeVoidMethodsFrom(Set<Method> candidates) {
|
||||
removeMethodsMatchingSelector(candidates, new MethodSelector() {
|
||||
public boolean select(Method method) {
|
||||
return method.getReturnType().getName().equals("void");
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
private Set<Method> removeUnanotatedFrom(Set<Method> candidates) {
|
||||
return removeMethodsMatchingSelector(candidates, new MethodSelector() {
|
||||
public boolean select(Method method) {
|
||||
Aggregator annotation = method.getAnnotation(Aggregator.class);
|
||||
return (annotation == null);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
private Set<Method> removeUnfittingFrom(Set<Method> candidates) {
|
||||
return removeMethodsMatchingSelector(candidates, new MethodSelector() {
|
||||
public boolean select(Method method) {
|
||||
Annotation[][] parameterAnnotations = method.getParameterAnnotations();
|
||||
Class<?>[] parameterTypes = method.getParameterTypes();
|
||||
return (!isFittinglyAnnotated(parameterTypes, parameterAnnotations));
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
private boolean isFittinglyAnnotated(Class<?>[] parameterTypes, Annotation[][] parameterAnnotations) {
|
||||
int candidateParametersFound = 0;
|
||||
for (int i = 0; i < parameterTypes.length; i++) {
|
||||
Class<?> parameterType = parameterTypes[i];
|
||||
if (parameterType.isAssignableFrom(List.class)) {
|
||||
boolean headerAnnotationFound = false;
|
||||
for (Annotation annotation : parameterAnnotations[i]) {
|
||||
if (annotation instanceof Header) {
|
||||
headerAnnotationFound = true;
|
||||
}
|
||||
}
|
||||
if (!headerAnnotationFound) {
|
||||
candidateParametersFound++;
|
||||
}
|
||||
}
|
||||
}
|
||||
return candidateParametersFound == 1;
|
||||
}
|
||||
|
||||
private void removeObjectMethodsFrom(Set<Method> candidates) {
|
||||
removeMethodsMatchingSelector(candidates, new MethodSelector() {
|
||||
public boolean select(Method method) {
|
||||
return method.getDeclaringClass().equals(Object.class);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
public void processAndSend(Object correlationKey,
|
||||
Collection<Message<?>> messagesUpForProcessing,
|
||||
MessageChannel outputChannel,
|
||||
BufferedMessagesCallback processedCallback) {
|
||||
Message reply = MessageBuilder.withPayload(
|
||||
this.adapter.executeMethod(messagesUpForProcessing)).build();
|
||||
processedCallback.onCompletionOf(correlationKey);
|
||||
processedCallback.onProcessingOf(messagesUpForProcessing
|
||||
.toArray(new Message[]{}));
|
||||
outputChannel.send(reply);
|
||||
}
|
||||
|
||||
public Set<Method> removeMethodsMatchingSelector(Set<Method> candidates, MethodSelector selector) {
|
||||
Set<Method> removed = new HashSet<Method>();
|
||||
Iterator<Method> iterator = candidates.iterator();
|
||||
while (iterator.hasNext()) {
|
||||
Method method = iterator.next();
|
||||
if (selector.select(method)) {
|
||||
iterator.remove();
|
||||
removed.add(method);
|
||||
}
|
||||
}
|
||||
return removed;
|
||||
}
|
||||
|
||||
|
||||
private interface MethodSelector {
|
||||
boolean select(Method method);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -19,10 +19,7 @@ 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.mockito.runners.MockitoJUnitRunner;
|
||||
import org.springframework.integration.core.Message;
|
||||
import org.springframework.integration.core.MessageChannel;
|
||||
import org.springframework.integration.core.MessageHeaders;
|
||||
@@ -32,10 +29,12 @@ import org.springframework.integration.store.MessageStore;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
|
||||
import static org.mockito.Mockito.*;
|
||||
|
||||
/**
|
||||
* @author Iwein Fuld
|
||||
*/
|
||||
@RunWith(MockitoJUnit44Runner.class)
|
||||
@RunWith(MockitoJUnitRunner.class)
|
||||
public class BufferingMessageHandlerTest {
|
||||
|
||||
private BufferingMessageHandler buffer;
|
||||
|
||||
@@ -0,0 +1,167 @@
|
||||
package org.springframework.integration.aggregator;
|
||||
|
||||
import org.junit.Before;
|
||||
import org.junit.Test;
|
||||
import org.junit.runner.RunWith;
|
||||
import org.mockito.ArgumentCaptor;
|
||||
import org.mockito.Mock;
|
||||
import org.mockito.runners.MockitoJUnitRunner;
|
||||
import org.springframework.integration.annotation.Aggregator;
|
||||
import org.springframework.integration.annotation.Header;
|
||||
import org.springframework.integration.core.Message;
|
||||
import org.springframework.integration.core.MessageChannel;
|
||||
import org.springframework.integration.message.MessageBuilder;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collection;
|
||||
import java.util.List;
|
||||
|
||||
import static org.hamcrest.CoreMatchers.is;
|
||||
import static org.junit.Assert.assertThat;
|
||||
import static org.junit.Assert.fail;
|
||||
import static org.mockito.Matchers.isA;
|
||||
import static org.mockito.Mockito.verify;
|
||||
import static org.mockito.Mockito.when;
|
||||
|
||||
@RunWith(MockitoJUnitRunner.class)
|
||||
public class MethodInvokingMessagesProcessorTests {
|
||||
|
||||
@Mock
|
||||
private BufferedMessagesCallback processedCallback;
|
||||
|
||||
@Mock
|
||||
private MessageChannel outputChannel;
|
||||
|
||||
private Collection<Message<?>> messagesUpForProcessing = new ArrayList<Message<?>>(
|
||||
3);
|
||||
|
||||
@Before
|
||||
public void initializeMessagesUpForProcessing() {
|
||||
messagesUpForProcessing.add(MessageBuilder.withPayload(1).build());
|
||||
messagesUpForProcessing.add(MessageBuilder.withPayload(2).build());
|
||||
messagesUpForProcessing.add(MessageBuilder.withPayload(4).build());
|
||||
}
|
||||
|
||||
private class AnnotatedAggregatorMethod {
|
||||
|
||||
@Aggregator
|
||||
@SuppressWarnings("unused")
|
||||
public Integer and(List<Integer> flags) {
|
||||
int result = 0;
|
||||
for (Integer flag : flags) {
|
||||
result = result | flag;
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
public String know() {
|
||||
return "I'm not the one ";
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void shouldFindAnnotatedAggregatorMethod() throws Exception {
|
||||
MessagesProcessor processor = new MethodInvokingMessagesProcessor(
|
||||
new AnnotatedAggregatorMethod());
|
||||
ArgumentCaptor<Message> messageCaptor = ArgumentCaptor
|
||||
.forClass(Message.class);
|
||||
when(outputChannel.send(isA(Message.class))).thenReturn(true);
|
||||
processor.processAndSend(3, messagesUpForProcessing, outputChannel,
|
||||
processedCallback);
|
||||
// verify
|
||||
verify(outputChannel).send(messageCaptor.capture());
|
||||
assertThat((Integer) messageCaptor.getValue().getPayload(), is(7));
|
||||
}
|
||||
|
||||
|
||||
private class SimpleAggregator {
|
||||
public Integer and(List<Integer> flags) {
|
||||
int result = 0;
|
||||
for (Integer flag : flags) {
|
||||
result = result | flag;
|
||||
}
|
||||
return result;
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void shouldFindSimpleAggregatorMethod() throws Exception {
|
||||
MessagesProcessor processor = new MethodInvokingMessagesProcessor(
|
||||
new SimpleAggregator());
|
||||
ArgumentCaptor<Message> messageCaptor = ArgumentCaptor
|
||||
.forClass(Message.class);
|
||||
when(outputChannel.send(isA(Message.class))).thenReturn(true);
|
||||
processor.processAndSend(3, messagesUpForProcessing, outputChannel,
|
||||
processedCallback);
|
||||
// verify
|
||||
verify(outputChannel).send(messageCaptor.capture());
|
||||
assertThat((Integer) messageCaptor.getValue().getPayload(), is(7));
|
||||
}
|
||||
|
||||
|
||||
private class UnnanotatedAggregator {
|
||||
public Integer and(List<Integer> flags) {
|
||||
int result = 0;
|
||||
for (Integer flag : flags) {
|
||||
result = result | flag;
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
public void voidMethodShouldBeIgnored(List<Integer> flags){
|
||||
fail("this method should not be invoked");
|
||||
}
|
||||
public String methodAcceptingNoCollectionShouldBeIgnored(@Header String irrelevant){
|
||||
fail("this method should not be invoked");
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void shouldFindFittingMethodAmongMultipleUnanotated() {
|
||||
MessagesProcessor processor = new MethodInvokingMessagesProcessor(
|
||||
new UnnanotatedAggregator()
|
||||
);
|
||||
|
||||
ArgumentCaptor<Message> messageCaptor = ArgumentCaptor
|
||||
.forClass(Message.class) ;
|
||||
|
||||
when(outputChannel.send(isA(Message.class))).thenReturn(true);
|
||||
processor.processAndSend(3, messagesUpForProcessing, outputChannel,
|
||||
processedCallback);
|
||||
// verify
|
||||
verify(outputChannel).send(messageCaptor.capture());
|
||||
assertThat((Integer) messageCaptor.getValue().getPayload(), is(7));
|
||||
}
|
||||
private class AnnotatedParametersAggregator {
|
||||
public Integer and(List<Integer> flags) {
|
||||
int result = 0;
|
||||
for (Integer flag : flags) {
|
||||
result = result | flag;
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
public String listHeaderShouldBeIgnored(@Header List<Integer> flags){
|
||||
fail("this method should not be invoked");
|
||||
return "";
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void shouldFindFittingMethodAmongMultipleWithAnnotatedParameters() {
|
||||
MessagesProcessor processor = new MethodInvokingMessagesProcessor(
|
||||
new AnnotatedParametersAggregator()
|
||||
);
|
||||
|
||||
ArgumentCaptor<Message> messageCaptor = ArgumentCaptor
|
||||
.forClass(Message.class) ;
|
||||
|
||||
when(outputChannel.send(isA(Message.class))).thenReturn(true);
|
||||
processor.processAndSend(3, messagesUpForProcessing, outputChannel,
|
||||
processedCallback);
|
||||
// verify
|
||||
verify(outputChannel).send(messageCaptor.capture());
|
||||
assertThat((Integer) messageCaptor.getValue().getPayload(), is(7));
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user