diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/BufferingMessageHandler.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/BufferingMessageHandler.java index 082492ee17..33cbb9e9f5 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/BufferingMessageHandler.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/BufferingMessageHandler.java @@ -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 keysInBuffer = new DelayQueue(); private volatile long timeout = 60000l; diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageListMethodAdapter.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageListMethodAdapter.java index 366f2377fa..43118c75e8 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageListMethodAdapter.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageListMethodAdapter.java @@ -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> 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> 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> messages) { - List payloadList = new ArrayList(); - for (Message message : messages) { - payloadList.add(message.getPayload()); - } - return payloadList; - } + private List extractPayloadsFromMessages(Collection> messages) { + List payloadList = new ArrayList(); + 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; + } } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MethodInvokingAggregatorFactoryBean.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MethodInvokingAggregatorFactoryBean.java new file mode 100644 index 0000000000..126c4f0d11 --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MethodInvokingAggregatorFactoryBean.java @@ -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 , 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 getObjectType() { + return BufferingMessageHandler.class; + } + + public boolean isSingleton() { + return false; + } +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MethodInvokingMessagesProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MethodInvokingMessagesProcessor.java new file mode 100644 index 0000000000..299681cd96 --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MethodInvokingMessagesProcessor.java @@ -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 candidates = new HashSet(Arrays.asList(methods)); + + removeObjectMethodsFrom(candidates); + removeVoidMethodsFrom(candidates); + removeListIncompatibleMethodsFrom(candidates); + Set unanotatedCandidates = new HashSet(); + 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 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 candidates) { + removeMethodsMatchingSelector(candidates, new MethodSelector() { + public boolean select(Method method) { + return method.getReturnType().getName().equals("void"); + } + }); + } + + private Set removeUnanotatedFrom(Set candidates) { + return removeMethodsMatchingSelector(candidates, new MethodSelector() { + public boolean select(Method method) { + Aggregator annotation = method.getAnnotation(Aggregator.class); + return (annotation == null); + } + }); + } + + private Set removeUnfittingFrom(Set 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 candidates) { + removeMethodsMatchingSelector(candidates, new MethodSelector() { + public boolean select(Method method) { + return method.getDeclaringClass().equals(Object.class); + } + }); + } + + public void processAndSend(Object correlationKey, + Collection> 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 removeMethodsMatchingSelector(Set candidates, MethodSelector selector) { + Set removed = new HashSet(); + Iterator 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); + } + +} 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 index 277a5bc2a0..272eaf969b 100644 --- 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 @@ -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; diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/MethodInvokingMessagesProcessorTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/MethodInvokingMessagesProcessorTests.java new file mode 100644 index 0000000000..c0ee627cdd --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/MethodInvokingMessagesProcessorTests.java @@ -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> messagesUpForProcessing = new ArrayList>( + 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 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 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 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 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 flags) { + int result = 0; + for (Integer flag : flags) { + result = result | flag; + } + return result; + } + + public void voidMethodShouldBeIgnored(List 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 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 flags) { + int result = 0; + for (Integer flag : flags) { + result = result | flag; + } + return result; + } + + public String listHeaderShouldBeIgnored(@Header List flags){ + fail("this method should not be invoked"); + return ""; + } + } + + @Test + public void shouldFindFittingMethodAmongMultipleWithAnnotatedParameters() { + MessagesProcessor processor = new MethodInvokingMessagesProcessor( + new AnnotatedParametersAggregator() + ); + + ArgumentCaptor 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)); + } +}