From a5c14782fbd066feea80aaec9a2786d7617ebaad Mon Sep 17 00:00:00 2001 From: David Syer Date: Thu, 29 Jul 2010 06:26:56 +0000 Subject: [PATCH] INT-1285: @ReleaseStrategy and @CorrelationStrategy working in XML config --- .../CorrelationStrategyAdapter.java | 2 +- .../aggregator/MessageListMethodAdapter.java | 27 +-- .../MessageListMethodAdapterHelper.java | 160 ++++++++++++++++ .../MethodInvokingMessageGroupProcessor.java | 143 ++------------- .../aggregator/ReleaseStrategyAdapter.java | 17 +- .../integration/config/AnnotationFinder.java | 61 +++++++ .../CorrelationStrategyFactoryBean.java | 83 +++++++++ .../config/ReleaseStrategyFactoryBean.java | 82 +++++++++ .../AggregatorAnnotationPostProcessor.java | 6 +- .../config/xml/AggregatorParser.java | 172 +++++++++--------- .../CorrelationStrategyAdapterTests.java | 2 +- ...hodInvokingMessageGroupProcessorTests.java | 4 +- .../AggregatorIntegrationTests.java | 14 +- .../AnnotationAggregatorTests-context.xml | 18 ++ .../AnnotationAggregatorTests.java | 80 ++++++++ ...aultMessageAggregatorIntegrationTests.java | 18 +- ...vokingAggregatorReturningMessageTests.java | 1 - .../config/AggregatorParserTests.java | 36 ++-- .../annotation/AggregatorAnnotationTests.java | 64 +++---- 19 files changed, 686 insertions(+), 304 deletions(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/aggregator/MessageListMethodAdapterHelper.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/config/AnnotationFinder.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/config/CorrelationStrategyFactoryBean.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/config/ReleaseStrategyFactoryBean.java create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AnnotationAggregatorTests-context.xml create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AnnotationAggregatorTests.java diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/CorrelationStrategyAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/CorrelationStrategyAdapter.java index 7cbaaf5e28..c5ecaee8f0 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/CorrelationStrategyAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/CorrelationStrategyAdapter.java @@ -26,12 +26,12 @@ import org.springframework.util.Assert; * {@link CorrelationStrategy} implementation that works as an adapter to another bean. * * @author Marius Bogoevici + * @author Dave Syer */ public class CorrelationStrategyAdapter implements CorrelationStrategy { private final MethodInvokingMessageProcessor processor; - public CorrelationStrategyAdapter(Object object, String methodName) { this.processor = new MethodInvokingMessageProcessor(object, methodName, true); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MessageListMethodAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MessageListMethodAdapter.java index 1cbf159411..ecc3fded5f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MessageListMethodAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MessageListMethodAdapter.java @@ -16,12 +16,6 @@ package org.springframework.integration.aggregator; -import org.springframework.integration.Message; -import org.springframework.integration.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; @@ -30,11 +24,19 @@ import java.util.ArrayList; import java.util.Collection; import java.util.List; +import org.springframework.integration.Message; +import org.springframework.integration.MessagingException; +import org.springframework.integration.util.DefaultMethodInvoker; +import org.springframework.util.Assert; +import org.springframework.util.ReflectionUtils; + /** * Base class for implementing adapters for methods which take as an argument a * list of {@link Message Message} instances or payloads. * * @author Marius Bogoevici + * @author Iwein Fuld + * @author Dave Syer */ public class MessageListMethodAdapter { @@ -42,7 +44,6 @@ public class MessageListMethodAdapter { 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"); @@ -67,12 +68,7 @@ public class MessageListMethodAdapter { return method; } - private static boolean isActualTypeParameterizedMessage(Method method) { - return (getCollectionActualType(method) instanceof ParameterizedType) - && Message.class.isAssignableFrom((Class) ((ParameterizedType) getCollectionActualType(method)).getRawType()); - } - - protected final Object executeMethod(Collection> messages) { + public final Object executeMethod(Collection> messages) { try { if (isMethodParameterParameterized(this.method) && isHavingActualTypeArguments(this.method) && (isActualTypeRawMessage(this.method) || isActualTypeParameterizedMessage(this.method))) { @@ -89,6 +85,11 @@ public class MessageListMethodAdapter { } } + private static boolean isActualTypeParameterizedMessage(Method method) { + return (getCollectionActualType(method) instanceof ParameterizedType) + && Message.class.isAssignableFrom((Class) ((ParameterizedType) getCollectionActualType(method)).getRawType()); + } + private List extractPayloadsFromMessages(Collection> messages) { List payloadList = new ArrayList(); for (Message message : messages) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MessageListMethodAdapterHelper.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MessageListMethodAdapterHelper.java new file mode 100644 index 0000000000..75d09a65c9 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MessageListMethodAdapterHelper.java @@ -0,0 +1,160 @@ +/* + * Copyright 2002-2010 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 java.lang.annotation.Annotation; +import java.lang.reflect.Method; +import java.util.Collection; +import java.util.HashSet; +import java.util.Iterator; +import java.util.Set; +import java.util.concurrent.atomic.AtomicReference; + +import org.springframework.aop.support.AopUtils; +import org.springframework.core.annotation.AnnotationUtils; +import org.springframework.integration.annotation.Header; +import org.springframework.util.Assert; +import org.springframework.util.ReflectionUtils; + +/** + * Convenience helper that looks for an appropriate method handling a list of messages and returns null if not found. + * + * @author Dave Syer + * + * @since 2.0 + */ +public class MessageListMethodAdapterHelper { + + public MessageListMethodAdapter getAdapter(Object candidate, Class annotationType) { + Method method = findAggregatorMethod(candidate, annotationType); + if (method == null) { + return null; + } + return new MessageListMethodAdapter(candidate, method); + } + + public Method findAggregatorMethod(Object candidate, Class annotationType) { + Class targetClass = AopUtils.getTargetClass(candidate); + if (targetClass == null) { + targetClass = candidate.getClass(); + } + Method method = this.findAnnotatedMethod(targetClass, annotationType); + if (method == null) { + method = this.findSinglePublicMethod(targetClass); + } + return method; + } + + private Method findAnnotatedMethod(final Class targetClass, final Class annotationType) { + final AtomicReference annotatedMethod = new AtomicReference(); + ReflectionUtils.doWithMethods(targetClass, new ReflectionUtils.MethodCallback() { + public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException { + Annotation annotation = AnnotationUtils.findAnnotation(method, annotationType); + if (annotation != null) { + Assert.isNull(annotatedMethod.get(), "found more than one method on target class [" + targetClass + + "] with the annotation type [" + annotationType.getName() + "]"); + annotatedMethod.set(method); + } + } + }); + return annotatedMethod.get(); + } + + private Method findSinglePublicMethod(Class targetClass) { + Set methods = new HashSet(); + for (Method method : targetClass.getMethods()) { + if (!method.getDeclaringClass().equals(Object.class)) { + methods.add(method); + } + } + removeListIncompatibleMethodsFrom(methods); + removeVoidMethodsFrom(methods); + removeUnfittingFrom(methods); + if (methods.size() > 1) { + throw new IllegalArgumentException("Class [" + targetClass + "] contains more than one public Method."); + } + return methods.isEmpty() ? null : methods.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 (Collection.class.isAssignableFrom(parameterClass)) { + 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 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 (Collection.class.isAssignableFrom(parameterType)) { + boolean headerAnnotationFound = false; + for (Annotation annotation : parameterAnnotations[i]) { + if (annotation instanceof Header) { + headerAnnotationFound = true; + } + } + if (!headerAnnotationFound) { + candidateParametersFound++; + } + } + } + return candidateParametersFound == 1; + } + + private 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/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessor.java index c376251a7b..93ccfda272 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessor.java @@ -16,28 +16,20 @@ package org.springframework.integration.aggregator; -import java.lang.annotation.Annotation; import java.lang.reflect.Method; import java.util.Collection; -import java.util.HashSet; -import java.util.Iterator; -import java.util.Set; -import java.util.concurrent.atomic.AtomicReference; -import org.springframework.aop.support.AopUtils; -import org.springframework.core.annotation.AnnotationUtils; import org.springframework.integration.Message; import org.springframework.integration.annotation.Aggregator; -import org.springframework.integration.annotation.Header; import org.springframework.integration.store.MessageGroup; import org.springframework.util.Assert; -import org.springframework.util.ReflectionUtils; /** * MessageGroupProcessor that serves as an adapter for the invocation of a POJO method. * * @author Iwein Fuld * @author Mark Fisher + * @author Dave Syer * @since 2.0 */ public class MethodInvokingMessageGroupProcessor extends AbstractAggregatingMessageGroupProcessor { @@ -45,14 +37,14 @@ public class MethodInvokingMessageGroupProcessor extends AbstractAggregatingMess private final MessageListMethodAdapter adapter; /** - * Creates a wrapper around the target passed in. This constructor will choose the best fitting method and throw an - * exception when methods are ambiguous or no fitting methods can be found. + * Creates a wrapper around the object passed in. This constructor will look for a method that can process + * a list of messages. * * @param target the object to wrap - * @throws IllegalStateException when no single method can be found unambiguously */ public MethodInvokingMessageGroupProcessor(Object target) { - this.adapter = new MessageListMethodAdapter(target, this.findAggregatorMethod(target)); + this.adapter = new MessageListMethodAdapterHelper().getAdapter(target, Aggregator.class); + Assert.notNull(this.adapter, "No aggregator method could be found for object of type: "+target.getClass()); } /** @@ -60,9 +52,19 @@ public class MethodInvokingMessageGroupProcessor extends AbstractAggregatingMess * fail when it cannot find a method with the given name. * * @param target the object to wrap - * @param method the name of the method to look for + * @param methodName the name of the method to invoke */ - public MethodInvokingMessageGroupProcessor(Object target, String method) { + public MethodInvokingMessageGroupProcessor(Object target, String methodName) { + this.adapter = new MessageListMethodAdapter(target, methodName); + } + + /** + * Creates a wrapper around the object passed in. + * + * @param target the object to wrap + * @param method the method to invoke + */ + public MethodInvokingMessageGroupProcessor(Object target, Method method) { this.adapter = new MessageListMethodAdapter(target, method); } @@ -73,115 +75,4 @@ public class MethodInvokingMessageGroupProcessor extends AbstractAggregatingMess return result; } - private Method findAggregatorMethod(Object candidate) { - Class targetClass = AopUtils.getTargetClass(candidate); - if (targetClass == null) { - targetClass = candidate.getClass(); - } - Method method = this.findAnnotatedMethod(targetClass); - if (method == null) { - method = this.findSinglePublicMethod(targetClass); - } - return method; - } - - private Method findAnnotatedMethod(final Class targetClass) { - final AtomicReference annotatedMethod = new AtomicReference(); - ReflectionUtils.doWithMethods(targetClass, new ReflectionUtils.MethodCallback() { - public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException { - Annotation annotation = AnnotationUtils.findAnnotation(method, Aggregator.class); - if (annotation != null) { - Assert.isNull(annotatedMethod.get(), "found more than one method on target class [" + targetClass - + "] with the annotation type [" + Aggregator.class.getName() + "]"); - annotatedMethod.set(method); - } - } - }); - return annotatedMethod.get(); - } - - private Method findSinglePublicMethod(Class targetClass) { - Set methods = new HashSet(); - for (Method method : targetClass.getMethods()) { - if (!method.getDeclaringClass().equals(Object.class)) { - methods.add(method); - } - } - removeListIncompatibleMethodsFrom(methods); - removeVoidMethodsFrom(methods); - removeUnfittingFrom(methods); - if (methods.size() > 1) { - throw new IllegalArgumentException("Class [" + targetClass + "] contains more than one public Method."); - } - return methods.isEmpty() ? null : methods.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 (Collection.class.isAssignableFrom(parameterClass)) { - 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 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 (Collection.class.isAssignableFrom(parameterType)) { - boolean headerAnnotationFound = false; - for (Annotation annotation : parameterAnnotations[i]) { - if (annotation instanceof Header) { - headerAnnotationFound = true; - } - } - if (!headerAnnotationFound) { - candidateParametersFound++; - } - } - } - return candidateParametersFound == 1; - } - - private 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/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ReleaseStrategyAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ReleaseStrategyAdapter.java index 0165417e44..36cf75bfbe 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ReleaseStrategyAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ReleaseStrategyAdapter.java @@ -28,28 +28,31 @@ import org.springframework.util.Assert; * attribute (e.g. <release-strategy ref="beanReference" method="methodName"/>). * * @author Marius Bogoevici + * @author Dave Syer */ -public class ReleaseStrategyAdapter extends MessageListMethodAdapter implements ReleaseStrategy { +public class ReleaseStrategyAdapter implements ReleaseStrategy { + + private final MessageListMethodAdapter adapter; public ReleaseStrategyAdapter(Object object, Method method) { - super(object, method); + adapter = new MessageListMethodAdapter(object, method); this.assertMethodReturnsBoolean(); } public ReleaseStrategyAdapter(Object object, String methodName) { - super(object, methodName); + adapter = new MessageListMethodAdapter(object, methodName); this.assertMethodReturnsBoolean(); } public boolean canRelease(MessageGroup messages) { - return ((Boolean) executeMethod(messages.getUnmarked())).booleanValue() && messages.getMarked().isEmpty(); + return ((Boolean) adapter.executeMethod(messages.getUnmarked())).booleanValue() && messages.getMarked().isEmpty(); } private void assertMethodReturnsBoolean() { - Assert.isTrue(Boolean.class.equals(this.getMethod().getReturnType()) - || boolean.class.equals(this.getMethod().getReturnType()), - "Method '" + getMethod().getName() + "' does not return a boolean value"); + Assert.isTrue(Boolean.class.equals(adapter.getMethod().getReturnType()) + || boolean.class.equals(adapter.getMethod().getReturnType()), + "Method '" + adapter.getMethod().getName() + "' does not return a boolean value"); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/AnnotationFinder.java b/spring-integration-core/src/main/java/org/springframework/integration/config/AnnotationFinder.java new file mode 100644 index 0000000000..635653f2c8 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/AnnotationFinder.java @@ -0,0 +1,61 @@ +/* + * Copyright 2002-2010 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.config; + +import java.lang.annotation.Annotation; +import java.lang.reflect.Method; +import java.util.concurrent.atomic.AtomicReference; + +import org.springframework.aop.support.AopUtils; +import org.springframework.core.annotation.AnnotationUtils; +import org.springframework.util.ReflectionUtils; +import org.springframework.util.ReflectionUtils.MethodCallback; + +/** + * Helper to provide common features for inspecting objects and locating annotated methods. + * + * @author Dave Syer + * + */ +class AnnotationFinder { + + public static Method findAnnotatedMethod(Object target, final Class annotationType) { + final AtomicReference reference = new AtomicReference(); + ReflectionUtils.doWithMethods(getTargetClass(target), new MethodCallback() { + public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException { + if (AnnotationUtils.findAnnotation(method, annotationType) != null) { + reference.set(method); + } + } + }); + return reference.get(); + } + + private static Class getTargetClass(Object targetObject) { + Class targetClass = targetObject.getClass(); + if (AopUtils.isAopProxy(targetObject)) { + targetClass = AopUtils.getTargetClass(targetObject); + } + else if (AopUtils.isCglibProxyClass(targetClass)) { + Class superClass = targetObject.getClass().getSuperclass(); + if (!Object.class.equals(superClass)) { + targetClass = superClass; + } + } + return targetClass; + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/CorrelationStrategyFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/CorrelationStrategyFactoryBean.java new file mode 100644 index 0000000000..f9a520c55a --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/CorrelationStrategyFactoryBean.java @@ -0,0 +1,83 @@ +/* + * Copyright 2002-2010 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.config; + +import java.lang.reflect.Method; + +import org.springframework.beans.factory.FactoryBean; +import org.springframework.integration.MessageHeaders; +import org.springframework.integration.aggregator.CorrelationStrategy; +import org.springframework.integration.aggregator.CorrelationStrategyAdapter; +import org.springframework.integration.aggregator.HeaderAttributeCorrelationStrategy; +import org.springframework.util.StringUtils; + +/** + * Convenience factory for XML configuration of a {@link CorrelationStrategy}. Encapsulates the knowledge of the default + * strategy and search algorithms for POJO and annotated methods. + * + * @author Dave Syer + * + */ +public class CorrelationStrategyFactoryBean implements FactoryBean { + + private CorrelationStrategy delegate = new HeaderAttributeCorrelationStrategy(MessageHeaders.CORRELATION_ID); + + /** + * Create a factory and set up the delegate which clients of the factory will see as its product. + * + * @param target the target object (null if default strategy is acceptable) + */ + public CorrelationStrategyFactoryBean(Object target) { + this(target, null); + } + + /** + * Create a factory and set up the delegate which clients of the factory will see as its product. + * + * @param target the target object (null if default strategy is acceptable) + * @param methodName the method name to invoke in the target (null if it can be inferred) + */ + public CorrelationStrategyFactoryBean(Object target, String methodName) { + if (target instanceof CorrelationStrategy && !StringUtils.hasText(methodName)) { + delegate = (CorrelationStrategy) target; + return; + } + if (target != null) { + if (StringUtils.hasText(methodName)) { + delegate = new CorrelationStrategyAdapter(target, methodName); + } + else { + Method method = AnnotationFinder.findAnnotatedMethod(target, org.springframework.integration.annotation.CorrelationStrategy.class); + if (method != null) { + delegate = new CorrelationStrategyAdapter(target, method); + } + } + } + } + + public CorrelationStrategy getObject() throws Exception { + return delegate; + } + + public Class getObjectType() { + return CorrelationStrategy.class; + } + + public boolean isSingleton() { + return true; + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/ReleaseStrategyFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/ReleaseStrategyFactoryBean.java new file mode 100644 index 0000000000..83ddca947c --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/ReleaseStrategyFactoryBean.java @@ -0,0 +1,82 @@ +/* + * Copyright 2002-2010 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.config; + +import java.lang.reflect.Method; + +import org.springframework.beans.factory.FactoryBean; +import org.springframework.integration.aggregator.ReleaseStrategy; +import org.springframework.integration.aggregator.ReleaseStrategyAdapter; +import org.springframework.integration.aggregator.SequenceSizeReleaseStrategy; +import org.springframework.util.StringUtils; + +/** + * Convenience factory for XML configuration of a {@link ReleaseStrategy}. Encapsulates the knowledge of the default + * strategy and search algorithms for POJO and annotated methods. + * + * @author Dave Syer + * + */ +public class ReleaseStrategyFactoryBean implements FactoryBean { + + private ReleaseStrategy delegate = new SequenceSizeReleaseStrategy(); + + /** + * Create a factory and set up the delegate which clients of the factory will see as its product. + * + * @param target the target object (null if default strategy is acceptable) + */ + public ReleaseStrategyFactoryBean(Object target) { + this(target, null); + } + + /** + * Create a factory and set up the delegate which clients of the factory will see as its product. + * + * @param target the target object (null if default strategy is acceptable) + * @param methodName the method name to invoke in the target (null if it can be inferred) + */ + public ReleaseStrategyFactoryBean(Object target, String methodName) { + if (target instanceof ReleaseStrategy && !StringUtils.hasText(methodName)) { + delegate = (ReleaseStrategy) target; + return; + } + if (target != null) { + if (StringUtils.hasText(methodName)) { + delegate = new ReleaseStrategyAdapter(target, methodName); + } + else { + Method method = AnnotationFinder.findAnnotatedMethod(target, org.springframework.integration.annotation.ReleaseStrategy.class); + if (method != null) { + delegate = new ReleaseStrategyAdapter(target, method); + } + } + } + } + + public ReleaseStrategy getObject() throws Exception { + return delegate; + } + + public Class getObjectType() { + return ReleaseStrategy.class; + } + + public boolean isSingleton() { + return true; + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AggregatorAnnotationPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AggregatorAnnotationPostProcessor.java index f7a642ad56..65ac190c54 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AggregatorAnnotationPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AggregatorAnnotationPostProcessor.java @@ -50,10 +50,10 @@ public class AggregatorAnnotationPostProcessor extends AbstractMethodAnnotationP @Override protected MessageHandler createHandler(Object bean, Method method, Aggregator annotation) { - MethodInvokingMessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor(bean, method.getName()); - ReleaseStrategyAdapter ReleaseStrategy = getReleaseStrategy(bean); + MethodInvokingMessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor(bean, method); + ReleaseStrategyAdapter releaseStrategy = getReleaseStrategy(bean); CorrelationStrategyAdapter correlationStrategy = getCorrelationStrategy(bean); - CorrelatingMessageHandler handler = new CorrelatingMessageHandler(processor, new SimpleMessageStore(), correlationStrategy, ReleaseStrategy); + CorrelatingMessageHandler handler = new CorrelatingMessageHandler(processor, new SimpleMessageStore(), correlationStrategy, releaseStrategy); String discardChannelName = annotation.discardChannel(); if (StringUtils.hasText(discardChannelName)) { MessageChannel discardChannel = this.channelResolver.resolveChannelName(discardChannelName); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AggregatorParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AggregatorParser.java index 79eda24103..8c4793ede4 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AggregatorParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AggregatorParser.java @@ -16,9 +16,10 @@ package org.springframework.integration.config.xml; +import org.springframework.beans.BeanMetadataElement; +import org.springframework.beans.factory.config.RuntimeBeanReference; import org.springframework.beans.factory.parsing.BeanComponentDefinition; import org.springframework.beans.factory.support.BeanDefinitionBuilder; -import org.springframework.beans.factory.support.BeanDefinitionReaderUtils; import org.springframework.beans.factory.xml.ParserContext; import org.springframework.util.StringUtils; import org.w3c.dom.Element; @@ -26,7 +27,7 @@ import org.w3c.dom.Element; /** * Parser for the aggregator element of the integration namespace. Registers the annotation-driven * post-processors. - * + * * @author Marius Bogoevici * @author Mark Fisher * @author Oleg Zhurakousky @@ -34,104 +35,107 @@ import org.w3c.dom.Element; */ public class AggregatorParser extends AbstractConsumerEndpointParser { - private static final String RELEASE_STRATEGY_REF_ATTRIBUTE = "release-strategy"; + private static final String RELEASE_STRATEGY_REF_ATTRIBUTE = "release-strategy"; - private static final String RELEASE_STRATEGY_METHOD_ATTRIBUTE = "release-strategy-method"; + private static final String RELEASE_STRATEGY_METHOD_ATTRIBUTE = "release-strategy-method"; - private static final String CORRELATION_STRATEGY_REF_ATTRIBUTE = "correlation-strategy"; + private static final String CORRELATION_STRATEGY_REF_ATTRIBUTE = "correlation-strategy"; - private static final String CORRELATION_STRATEGY_METHOD_ATTRIBUTE = "correlation-strategy-method"; + private static final String CORRELATION_STRATEGY_METHOD_ATTRIBUTE = "correlation-strategy-method"; - private static final String MESSAGE_STORE_ATTRIBUTE = "message-store"; + private static final String MESSAGE_STORE_ATTRIBUTE = "message-store"; - private static final String OUTPUT_CHANNEL_ATTRIBUTE = "output-channel"; + private static final String OUTPUT_CHANNEL_ATTRIBUTE = "output-channel"; - private static final String DISCARD_CHANNEL_ATTRIBUTE = "discard-channel"; + private static final String DISCARD_CHANNEL_ATTRIBUTE = "discard-channel"; - private static final String SEND_TIMEOUT_ATTRIBUTE = "send-timeout"; + private static final String SEND_TIMEOUT_ATTRIBUTE = "send-timeout"; - private static final String SEND_PARTIAL_RESULT_ON_EXPIRY_ATTRIBUTE = "send-partial-result-on-expiry"; + private static final String SEND_PARTIAL_RESULT_ON_EXPIRY_ATTRIBUTE = "send-partial-result-on-expiry"; - private static final String RELEASE_STRATEGY_PROPERTY = "releaseStrategy"; + private static final String RELEASE_STRATEGY_PROPERTY = "releaseStrategy"; - private static final String CORRELATION_STRATEGY_PROPERTY = "correlationStrategy"; + private static final String CORRELATION_STRATEGY_PROPERTY = "correlationStrategy"; + @Override + protected BeanDefinitionBuilder parseHandler(Element element, ParserContext parserContext) { + BeanComponentDefinition innerHandlerDefinition = IntegrationNamespaceUtils.parseInnerHandlerDefinition(element, + parserContext); + String ref = element.getAttribute(REF_ATTRIBUTE); + BeanDefinitionBuilder builder; - @Override - protected BeanDefinitionBuilder parseHandler(Element element, ParserContext parserContext) { - BeanComponentDefinition innerHandlerDefinition = IntegrationNamespaceUtils.parseInnerHandlerDefinition(element, parserContext); - String ref = element.getAttribute(REF_ATTRIBUTE); - BeanDefinitionBuilder builder; + builder = BeanDefinitionBuilder.genericBeanDefinition(IntegrationNamespaceUtils.BASE_PACKAGE + + ".aggregator.CorrelatingMessageHandler"); + BeanDefinitionBuilder processorBuilder = null; + BeanMetadataElement processor = null; - builder = BeanDefinitionBuilder.genericBeanDefinition( - IntegrationNamespaceUtils.BASE_PACKAGE + ".aggregator.CorrelatingMessageHandler"); - BeanDefinitionBuilder processorBuilder = null; + if (innerHandlerDefinition != null || StringUtils.hasText(ref)) { + processorBuilder = BeanDefinitionBuilder.genericBeanDefinition(IntegrationNamespaceUtils.BASE_PACKAGE + + ".aggregator.MethodInvokingMessageGroupProcessor"); + builder.addConstructorArgValue(processorBuilder.getBeanDefinition()); + if (innerHandlerDefinition != null) { + processor = innerHandlerDefinition; + } + else { + processor = new RuntimeBeanReference(ref); + } + processorBuilder.addConstructorArgValue(processor); + } + else { + builder.addConstructorArgValue(BeanDefinitionBuilder.genericBeanDefinition( + IntegrationNamespaceUtils.BASE_PACKAGE + ".aggregator.DefaultAggregatingMessageGroupProcessor") + .getBeanDefinition()); + } - if (innerHandlerDefinition != null || StringUtils.hasText(ref)) { - processorBuilder = BeanDefinitionBuilder.genericBeanDefinition( - IntegrationNamespaceUtils.BASE_PACKAGE + ".aggregator.MethodInvokingMessageGroupProcessor"); - builder.addConstructorArgValue(processorBuilder.getBeanDefinition()); - } else { - builder.addConstructorArgValue(BeanDefinitionBuilder.genericBeanDefinition( - IntegrationNamespaceUtils.BASE_PACKAGE + ".aggregator.DefaultAggregatingMessageGroupProcessor").getBeanDefinition()); - } + if (StringUtils.hasText(element.getAttribute(METHOD_ATTRIBUTE))) { + String method = element.getAttribute(METHOD_ATTRIBUTE); + processorBuilder.getRawBeanDefinition().getConstructorArgumentValues().addGenericArgumentValue(method, + "java.lang.String"); + } - if (innerHandlerDefinition != null) { - processorBuilder.addConstructorArgValue(innerHandlerDefinition); - } else { - if (StringUtils.hasText(ref)) { - processorBuilder.addConstructorArgReference(ref); - } - } - if (StringUtils.hasText(element.getAttribute(METHOD_ATTRIBUTE))) { - String method = element.getAttribute(METHOD_ATTRIBUTE); - processorBuilder.getRawBeanDefinition().getConstructorArgumentValues().addGenericArgumentValue(method, "java.lang.String"); - } + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, MESSAGE_STORE_ATTRIBUTE); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, DISCARD_CHANNEL_ATTRIBUTE); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, OUTPUT_CHANNEL_ATTRIBUTE); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, SEND_TIMEOUT_ATTRIBUTE); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, SEND_PARTIAL_RESULT_ON_EXPIRY_ATTRIBUTE); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-startup"); + this.injectPropertyWithBean(RELEASE_STRATEGY_REF_ATTRIBUTE, RELEASE_STRATEGY_METHOD_ATTRIBUTE, + RELEASE_STRATEGY_PROPERTY, "ReleaseStrategy", element, builder, processor, parserContext); + this + .injectPropertyWithBean(CORRELATION_STRATEGY_REF_ATTRIBUTE, CORRELATION_STRATEGY_METHOD_ATTRIBUTE, + CORRELATION_STRATEGY_PROPERTY, "CorrelationStrategy", element, builder, processor, + parserContext); + return builder; + } - IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, - MESSAGE_STORE_ATTRIBUTE); - IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, - DISCARD_CHANNEL_ATTRIBUTE); - IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, - OUTPUT_CHANNEL_ATTRIBUTE); - IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, - SEND_TIMEOUT_ATTRIBUTE); - IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, - SEND_PARTIAL_RESULT_ON_EXPIRY_ATTRIBUTE); - IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-startup"); - this.injectPropertyWithBean(RELEASE_STRATEGY_REF_ATTRIBUTE, - RELEASE_STRATEGY_METHOD_ATTRIBUTE, RELEASE_STRATEGY_PROPERTY, - "ReleaseStrategyAdapter", element, builder, parserContext); - this.injectPropertyWithBean(CORRELATION_STRATEGY_REF_ATTRIBUTE, - CORRELATION_STRATEGY_METHOD_ATTRIBUTE, CORRELATION_STRATEGY_PROPERTY, - "CorrelationStrategyAdapter", element, builder, parserContext); - return builder; - } + private void injectPropertyWithBean(String beanRefAttribute, String methodRefAttribute, String beanProperty, + String adapterClass, Element element, BeanDefinitionBuilder builder, + BeanMetadataElement processor, ParserContext parserContext) { + final String beanRef = element.getAttribute(beanRefAttribute); + final String beanMethod = element.getAttribute(methodRefAttribute); + BeanMetadataElement adapter = null; + if (StringUtils.hasText(beanRef)) { + adapter = this.createAdapter(new RuntimeBeanReference(beanRef), beanMethod, adapterClass, parserContext); + } + else if (processor != null) { + adapter = this.createAdapter(processor, beanMethod, adapterClass, parserContext); + } + else { + adapter = this.createAdapter(null, beanMethod, adapterClass, parserContext); + } + builder.addPropertyValue(beanProperty, adapter); + } - private void injectPropertyWithBean(String beanRefAttribute, String methodRefAttribute, - String beanProperty, String adapterClass, Element element, - BeanDefinitionBuilder builder, ParserContext parserContext) { - final String beanRef = element.getAttribute(beanRefAttribute); - final String beanMethod = element.getAttribute(methodRefAttribute); - if (StringUtils.hasText(beanRef)) { - if (StringUtils.hasText(beanMethod)) { - String adapterBeanName = this.createAdapter(beanRef, beanMethod, adapterClass, - parserContext); - builder.addPropertyReference(beanProperty, adapterBeanName); - } else { - builder.addPropertyReference(beanProperty, beanRef); - } - } - } - - private String createAdapter(String ref, String method, String unqualifiedClassName, - ParserContext parserContext) { - BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition( - IntegrationNamespaceUtils.BASE_PACKAGE + ".aggregator." + unqualifiedClassName); - builder.addConstructorArgReference(ref); - builder.getRawBeanDefinition().getConstructorArgumentValues().addGenericArgumentValue(method, "java.lang.String"); - return BeanDefinitionReaderUtils.registerWithGeneratedName(builder.getBeanDefinition(), - parserContext.getRegistry()); - } + private BeanMetadataElement createAdapter(BeanMetadataElement ref, String method, String unqualifiedClassName, + ParserContext parserContext) { + BeanDefinitionBuilder builder = BeanDefinitionBuilder + .genericBeanDefinition(IntegrationNamespaceUtils.BASE_PACKAGE + ".config." + unqualifiedClassName + + "FactoryBean"); + builder.addConstructorArgValue(ref); + if (StringUtils.hasText(method)) { + builder.addConstructorArgValue(method); + } + return builder.getBeanDefinition(); + } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelationStrategyAdapterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelationStrategyAdapterTests.java index 312aa767db..06ee9cc7ec 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelationStrategyAdapterTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelationStrategyAdapterTests.java @@ -38,7 +38,7 @@ public class CorrelationStrategyAdapterTests { } @Test - public void testCorrelationStrategyAdapterObjectString() { + public void testMethodName() { CorrelationStrategyAdapter adapter = new CorrelationStrategyAdapter(new SimpleMessageCorrelator(), "getKey"); assertEquals("b", adapter.getCorrelationKey(message)); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessorTests.java index 02f13044eb..b41eb78d18 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessorTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessorTests.java @@ -125,7 +125,7 @@ public class MethodInvokingMessageGroupProcessorTests { } @SuppressWarnings("unused") - private class UnnanotatedAggregator { + private class UnannotatedAggregator { public Integer and(List flags) { int result = 0; for (Integer flag : flags) { @@ -146,7 +146,7 @@ public class MethodInvokingMessageGroupProcessorTests { @Test public void shouldFindFittingMethodAmongMultipleUnannotated() { - MessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor(new UnnanotatedAggregator()); + MessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor(new UnannotatedAggregator()); @SuppressWarnings("unchecked") ArgumentCaptor messageCaptor = ArgumentCaptor.forClass(Message.class); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests.java index 72bbe7aeb7..2ad38d36ea 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests.java @@ -16,6 +16,12 @@ package org.springframework.integration.aggregator.integration; +import static org.junit.Assert.assertEquals; + +import java.util.HashMap; +import java.util.List; +import java.util.Map; + import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; @@ -27,12 +33,6 @@ import org.springframework.integration.core.PollableChannel; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; -import java.util.HashMap; -import java.util.List; -import java.util.Map; - -import static org.junit.Assert.assertEquals; - /** * @author Iwein Fuld * @author Alex Peters @@ -51,7 +51,7 @@ public class AggregatorIntegrationTests { private PollableChannel output; @Test(timeout=5000) - public void aggregate() throws Exception { + public void testVanillaAggregation() throws Exception { for (int i = 0; i < 5; i++) { Map headers = stubHeaders(i, 5, 1); input.send(new GenericMessage(i, headers)); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AnnotationAggregatorTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AnnotationAggregatorTests-context.xml new file mode 100644 index 0000000000..571d4d4d19 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AnnotationAggregatorTests-context.xml @@ -0,0 +1,18 @@ + + + + + + + + + + + + \ No newline at end of file diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AnnotationAggregatorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AnnotationAggregatorTests.java new file mode 100644 index 0000000000..522abd3a49 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AnnotationAggregatorTests.java @@ -0,0 +1,80 @@ +/* + * Copyright 2002-2010 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.integration; + +import static org.junit.Assert.assertTrue; + +import java.util.List; + +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.integration.Message; +import org.springframework.integration.annotation.Aggregator; +import org.springframework.integration.annotation.CorrelationStrategy; +import org.springframework.integration.annotation.ReleaseStrategy; +import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.core.MessageBuilder; +import org.springframework.integration.core.PollableChannel; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Mark Fisher + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +public class AnnotationAggregatorTests { + + @Autowired + DirectChannel input; + + @Autowired + PollableChannel output; + + + @Test + public void testAggregationWithAnnotationStrategies() { + input.send(MessageBuilder.withPayload("a").build()); + input.send(MessageBuilder.withPayload("b").build()); + @SuppressWarnings("unchecked") + Message result = (Message) output.receive(); + String payload = result.getPayload(); + assertTrue("Wrong payload: "+payload, payload.contains("Payload=a")); + assertTrue("Wrong payload: "+payload, payload.contains("Payload=b")); + } + + @SuppressWarnings("unused") + private static class TestAggregator { + + @Aggregator + public Message aggregate(final List> messages) { + return MessageBuilder.withPayload(messages.toString()).build(); + } + + @ReleaseStrategy + public boolean release(final List> messages) { + return messages.size()>1; + } + + @CorrelationStrategy + public Object getKey(Message message) { + return "1"; + } + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/DefaultMessageAggregatorIntegrationTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/DefaultMessageAggregatorIntegrationTests.java index cb0977e7fd..a50c977740 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/DefaultMessageAggregatorIntegrationTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/DefaultMessageAggregatorIntegrationTests.java @@ -16,6 +16,15 @@ package org.springframework.integration.aggregator.integration; +import static org.hamcrest.CoreMatchers.is; +import static org.junit.Assert.assertThat; +import static org.junit.Assert.assertTrue; + +import java.util.Arrays; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; @@ -27,15 +36,6 @@ import org.springframework.integration.core.PollableChannel; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; -import java.util.Arrays; -import java.util.HashMap; -import java.util.List; -import java.util.Map; - -import static org.hamcrest.CoreMatchers.is; -import static org.junit.Assert.assertThat; -import static org.junit.Assert.assertTrue; - /** * @author Alex Peters * @author Iwein Fuld diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/MethodInvokingAggregatorReturningMessageTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/MethodInvokingAggregatorReturningMessageTests.java index a67cccf18a..1f16c68c08 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/MethodInvokingAggregatorReturningMessageTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/MethodInvokingAggregatorReturningMessageTests.java @@ -23,7 +23,6 @@ import java.util.List; import org.junit.Test; import org.junit.runner.RunWith; - import org.springframework.beans.factory.annotation.Autowired; import org.springframework.integration.Message; import org.springframework.integration.channel.DirectChannel; diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests.java index b8c6100d16..56e9dde76f 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests.java @@ -16,6 +16,14 @@ package org.springframework.integration.config; +import static org.hamcrest.CoreMatchers.is; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertThat; + +import java.lang.reflect.Method; +import java.util.ArrayList; +import java.util.List; + import org.junit.Assert; import org.junit.Before; import org.junit.Test; @@ -24,7 +32,11 @@ import org.springframework.beans.factory.BeanCreationException; import org.springframework.context.ApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.Message; -import org.springframework.integration.aggregator.*; +import org.springframework.integration.aggregator.CorrelatingMessageHandler; +import org.springframework.integration.aggregator.CorrelationStrategy; +import org.springframework.integration.aggregator.MessageListMethodAdapter; +import org.springframework.integration.aggregator.ReleaseStrategy; +import org.springframework.integration.aggregator.ReleaseStrategyAdapter; import org.springframework.integration.core.MessageBuilder; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.core.PollableChannel; @@ -32,14 +44,6 @@ import org.springframework.integration.endpoint.EventDrivenConsumer; import org.springframework.integration.test.util.TestUtils; import org.springframework.integration.util.MethodInvoker; -import java.lang.reflect.Method; -import java.util.ArrayList; -import java.util.List; - -import static org.hamcrest.CoreMatchers.is; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertThat; - /** * @author Marius Bogoevici * @author Mark Fisher @@ -77,7 +81,7 @@ public class AggregatorParserTests { public void testPropertyAssignment() throws Exception { EventDrivenConsumer endpoint = (EventDrivenConsumer) context.getBean("completelyDefinedAggregator"); - ReleaseStrategy ReleaseStrategy = (ReleaseStrategy) context.getBean("releaseStrategy"); + ReleaseStrategy releaseStrategy = (ReleaseStrategy) context.getBean("releaseStrategy"); CorrelationStrategy correlationStrategy = (CorrelationStrategy) context.getBean("correlationStrategy"); MessageChannel outputChannel = (MessageChannel) context.getBean("outputChannel"); MessageChannel discardChannel = (MessageChannel) context.getBean("discardChannel"); @@ -89,7 +93,7 @@ public class AggregatorParserTests { expectedMethod, ((MessageListMethodAdapter) new DirectFieldAccessor(accessor.getPropertyValue("outputProcessor")).getPropertyValue("adapter")).getMethod()); assertEquals( "The AggregatorEndpoint is not injected with the appropriate ReleaseStrategy instance", - ReleaseStrategy, accessor.getPropertyValue("releaseStrategy")); + releaseStrategy, accessor.getPropertyValue("releaseStrategy")); assertEquals("The AggregatorEndpoint is not injected with the appropriate CorrelationStrategy instance", correlationStrategy, accessor.getPropertyValue("correlationStrategy")); Assert.assertEquals("The AggregatorEndpoint is not injected with the appropriate output channel", @@ -135,13 +139,13 @@ public class AggregatorParserTests { MessageChannel input = (MessageChannel) context.getBean("aggregatorWithPojoReleaseStrategyInput"); EventDrivenConsumer endpoint = (EventDrivenConsumer) context.getBean("aggregatorWithPojoReleaseStrategy"); - ReleaseStrategy ReleaseStrategy = (ReleaseStrategy) new DirectFieldAccessor( + ReleaseStrategy releaseStrategy = (ReleaseStrategy) new DirectFieldAccessor( new DirectFieldAccessor(endpoint).getPropertyValue("handler")).getPropertyValue("releaseStrategy"); - Assert.assertTrue(ReleaseStrategy instanceof ReleaseStrategyAdapter); - DirectFieldAccessor ReleaseStrategyAccessor = new DirectFieldAccessor(ReleaseStrategy); - MethodInvoker invoker = (MethodInvoker) ReleaseStrategyAccessor.getPropertyValue("invoker"); + Assert.assertTrue(releaseStrategy instanceof ReleaseStrategyAdapter); + DirectFieldAccessor releaseStrategyAccessor = new DirectFieldAccessor(new DirectFieldAccessor(releaseStrategy).getPropertyValue("adapter")); + MethodInvoker invoker = (MethodInvoker) releaseStrategyAccessor.getPropertyValue("invoker"); Assert.assertTrue(new DirectFieldAccessor(invoker).getPropertyValue("object") instanceof MaxValueReleaseStrategy); - Assert.assertTrue(((Method) ReleaseStrategyAccessor.getPropertyValue("method")).getName().equals("checkCompleteness")); + Assert.assertTrue(((Method) releaseStrategyAccessor.getPropertyValue("method")).getName().equals("checkCompleteness")); input.send(createMessage(1l, "correllationId", 4, 0, null)); input.send(createMessage(2l, "correllationId", 4, 1, null)); input.send(createMessage(3l, "correllationId", 4, 2, null)); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/AggregatorAnnotationTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/AggregatorAnnotationTests.java index a4f36344a4..01e1364d9e 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/AggregatorAnnotationTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/AggregatorAnnotationTests.java @@ -56,8 +56,8 @@ public class AggregatorAnnotationTests { assertTrue(getPropertyValue(aggregator, "releaseStrategy") instanceof SequenceSizeReleaseStrategy); assertNull(getPropertyValue(aggregator, "outputChannel")); assertTrue(getPropertyValue(aggregator, "discardChannel") instanceof NullChannel); - assertEquals(CorrelatingMessageHandler.DEFAULT_SEND_TIMEOUT, - getPropertyValue(aggregator, "messagingTemplate.sendTimeout")); + assertEquals(CorrelatingMessageHandler.DEFAULT_SEND_TIMEOUT, getPropertyValue(aggregator, + "messagingTemplate.sendTimeout")); assertEquals(false, getPropertyValue(aggregator, "sendPartialResultOnExpiry")); } @@ -67,13 +67,11 @@ public class AggregatorAnnotationTests { new String[] { "classpath:/org/springframework/integration/config/annotation/testAnnotatedAggregator.xml" }); final String endpointName = "endpointWithCustomizedAnnotation"; MessageHandler aggregator = this.getAggregator(context, endpointName); - assertTrue(getPropertyValue(aggregator, "releaseStrategy") - instanceof SequenceSizeReleaseStrategy); + assertTrue(getPropertyValue(aggregator, "releaseStrategy") instanceof SequenceSizeReleaseStrategy); ChannelResolver channelResolver = new BeanFactoryChannelResolver(context); - assertEquals(channelResolver.resolveChannelName("outputChannel"), - getPropertyValue(aggregator, "outputChannel")); - assertEquals(channelResolver.resolveChannelName("discardChannel"), - getPropertyValue(aggregator, "discardChannel")); + assertEquals(channelResolver.resolveChannelName("outputChannel"), getPropertyValue(aggregator, "outputChannel")); + assertEquals(channelResolver.resolveChannelName("discardChannel"), getPropertyValue(aggregator, + "discardChannel")); assertEquals(98765432l, getPropertyValue(aggregator, "messagingTemplate.sendTimeout")); assertEquals(true, getPropertyValue(aggregator, "sendPartialResultOnExpiry")); } @@ -86,40 +84,38 @@ public class AggregatorAnnotationTests { MessageHandler aggregator = this.getAggregator(context, endpointName); Object ReleaseStrategy = getPropertyValue(aggregator, "releaseStrategy"); Assert.assertTrue(ReleaseStrategy instanceof ReleaseStrategyAdapter); - ReleaseStrategyAdapter ReleaseStrategyAdapter = (ReleaseStrategyAdapter) ReleaseStrategy; - DirectFieldAccessor invokerAccessor = new DirectFieldAccessor( - new DirectFieldAccessor(ReleaseStrategyAdapter).getPropertyValue("invoker")); + ReleaseStrategyAdapter releaseStrategyAdapter = (ReleaseStrategyAdapter) ReleaseStrategy; + DirectFieldAccessor invokerAccessor = new DirectFieldAccessor(new DirectFieldAccessor(new DirectFieldAccessor( + releaseStrategyAdapter).getPropertyValue("adapter")).getPropertyValue("invoker")); Object targetObject = invokerAccessor.getPropertyValue("object"); assertSame(context.getBean(endpointName), targetObject); Method completionCheckerMethod = (Method) invokerAccessor.getPropertyValue("method"); assertEquals("completionChecker", completionCheckerMethod.getName()); } - @Test - public void testAnnotationWithCustomCorrelationStrategy() throws Exception { - ApplicationContext context = new ClassPathXmlApplicationContext( - new String[] { "classpath:/org/springframework/integration/config/annotation/testAnnotatedAggregator.xml" }); - final String endpointName = "endpointWithCorrelationStrategy"; - MessageHandler aggregator = this.getAggregator(context, endpointName); - Object correlationStrategy = getPropertyValue(aggregator, "correlationStrategy"); - Assert.assertTrue(correlationStrategy instanceof CorrelationStrategyAdapter); - CorrelationStrategyAdapter ReleaseStrategyAdapter = (CorrelationStrategyAdapter) correlationStrategy; - DirectFieldAccessor processorAccessor = new DirectFieldAccessor( - new DirectFieldAccessor(ReleaseStrategyAdapter).getPropertyValue("processor")); - Object targetObject = processorAccessor.getPropertyValue("targetObject"); - assertSame(context.getBean(endpointName), targetObject); - Map handlerMethods = (Map) processorAccessor.getPropertyValue("handlerMethods"); - assertEquals(1, handlerMethods.size()); - DirectFieldAccessor handlerMethodAccessor = new DirectFieldAccessor(handlerMethods.values().iterator().next()); - Method completionCheckerMethod = (Method) handlerMethodAccessor.getPropertyValue("method"); - assertEquals("correlate", completionCheckerMethod.getName()); - } - - + @Test + public void testAnnotationWithCustomCorrelationStrategy() throws Exception { + ApplicationContext context = new ClassPathXmlApplicationContext( + new String[] { "classpath:/org/springframework/integration/config/annotation/testAnnotatedAggregator.xml" }); + final String endpointName = "endpointWithCorrelationStrategy"; + MessageHandler aggregator = this.getAggregator(context, endpointName); + Object correlationStrategy = getPropertyValue(aggregator, "correlationStrategy"); + Assert.assertTrue(correlationStrategy instanceof CorrelationStrategyAdapter); + CorrelationStrategyAdapter ReleaseStrategyAdapter = (CorrelationStrategyAdapter) correlationStrategy; + DirectFieldAccessor processorAccessor = new DirectFieldAccessor(new DirectFieldAccessor(ReleaseStrategyAdapter) + .getPropertyValue("processor")); + Object targetObject = processorAccessor.getPropertyValue("targetObject"); + assertSame(context.getBean(endpointName), targetObject); + Map handlerMethods = (Map) processorAccessor.getPropertyValue("handlerMethods"); + assertEquals(1, handlerMethods.size()); + DirectFieldAccessor handlerMethodAccessor = new DirectFieldAccessor(handlerMethods.values().iterator().next()); + Method completionCheckerMethod = (Method) handlerMethodAccessor.getPropertyValue("method"); + assertEquals("correlate", completionCheckerMethod.getName()); + } private MessageHandler getAggregator(ApplicationContext context, final String endpointName) { - EventDrivenConsumer endpoint = (EventDrivenConsumer) context.getBean( - endpointName + ".aggregatingMethod.aggregator"); + EventDrivenConsumer endpoint = (EventDrivenConsumer) context.getBean(endpointName + + ".aggregatingMethod.aggregator"); return TestUtils.getPropertyValue(endpoint, "handler", MessageHandler.class); }