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

INT-1285: @ReleaseStrategy and @CorrelationStrategy working in XML config

Browse Source
This commit is contained in:
David Syer
2010-07-29 06:26:56 +00:00
parent e01d235e89
commit a5c14782fb
19 changed files with 686 additions and 304 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-integration-core/src/main/java/org/springframework/integration/aggregator/CorrelationStrategyAdapter.java
View File

@@ -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);
}

27
spring-integration-core/src/main/java/org/springframework/integration/aggregator/MessageListMethodAdapter.java
View File

@@ -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<? extends Message<?>> messages) {
public final Object executeMethod(Collection<? extends Message<?>> 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<? extends Message<?>> messages) {
List<Object> payloadList = new ArrayList<Object>();
for (Message<?> message : messages) {

160
spring-integration-core/src/main/java/org/springframework/integration/aggregator/MessageListMethodAdapterHelper.java Normal file
View File

@@ -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<? extends Annotation> annotationType) {
Method method = findAggregatorMethod(candidate, annotationType);
if (method == null) {
return null;
}
return new MessageListMethodAdapter(candidate, method);
}
public Method findAggregatorMethod(Object candidate, Class<? extends Annotation> 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<? extends Annotation> annotationType) {
final AtomicReference<Method> annotatedMethod = new AtomicReference<Method>();
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<Method> methods = new HashSet<Method>();
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<Method> 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<Method> candidates) {
removeMethodsMatchingSelector(candidates, new MethodSelector() {
public boolean select(Method method) {
return method.getReturnType().getName().equals("void");
}
});
}
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 (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<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);
}
}

143
spring-integration-core/src/main/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessor.java
View File

@@ -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<Method> annotatedMethod = new AtomicReference<Method>();
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<Method> methods = new HashSet<Method>();
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<Method> 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<Method> candidates) {
removeMethodsMatchingSelector(candidates, new MethodSelector() {
public boolean select(Method method) {
return method.getReturnType().getName().equals("void");
}
});
}
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 (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<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);
}
}

17
spring-integration-core/src/main/java/org/springframework/integration/aggregator/ReleaseStrategyAdapter.java
View File

@@ -28,28 +28,31 @@ import org.springframework.util.Assert;
* attribute (e.g. &lt;release-strategy ref="beanReference" method="methodName"/&gt;).
*
* @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");
}
}

61
spring-integration-core/src/main/java/org/springframework/integration/config/AnnotationFinder.java Normal file
View File

@@ -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<? extends Annotation> annotationType) {
final AtomicReference<Method> reference = new AtomicReference<Method>();
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;
}
}

83
spring-integration-core/src/main/java/org/springframework/integration/config/CorrelationStrategyFactoryBean.java Normal file
View File

@@ -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<CorrelationStrategy> {
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;
}
}

82
spring-integration-core/src/main/java/org/springframework/integration/config/ReleaseStrategyFactoryBean.java Normal file
View File

@@ -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<ReleaseStrategy> {
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;
}
}

6
spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AggregatorAnnotationPostProcessor.java
View File

@@ -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);

172
spring-integration-core/src/main/java/org/springframework/integration/config/xml/AggregatorParser.java
View File

@@ -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 <em>aggregator</em> 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();
}
}

2
spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelationStrategyAdapterTests.java
View File

@@ -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));
}

4
spring-integration-core/src/test/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessorTests.java
View File

@@ -125,7 +125,7 @@ public class MethodInvokingMessageGroupProcessorTests {
}
@SuppressWarnings("unused")
private class UnnanotatedAggregator {
private class UnannotatedAggregator {
public Integer and(List<Integer> 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<Message> messageCaptor = ArgumentCaptor.forClass(Message.class);

14
spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests.java
View File

@@ -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<String, Object> headers = stubHeaders(i, 5, 1);
input.send(new GenericMessage<Integer>(i, headers));

18
spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AnnotationAggregatorTests-context.xml Normal file
View File

@@ -0,0 +1,18 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns:beans="http://www.springframework.org/schema/beans"
xmlns="http://www.springframework.org/schema/integration"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration.xsd">
<channel id="output">
<queue/>
</channel>
<aggregator input-channel="input" output-channel="output" method="aggregate">
<beans:bean class="org.springframework.integration.aggregator.integration.AnnotationAggregatorTests$TestAggregator"/>
</aggregator>
</beans:beans>

80
spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AnnotationAggregatorTests.java Normal file
View File

@@ -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<String> result = (Message<String>) 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<Message<?>> messages) {
return MessageBuilder.withPayload(messages.toString()).build();
}
@ReleaseStrategy
public boolean release(final List<Message<?>> messages) {
return messages.size()>1;
}
@CorrelationStrategy
public Object getKey(Message<?> message) {
return "1";
}
}
}

18
spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/DefaultMessageAggregatorIntegrationTests.java
View File

@@ -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

1
spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/MethodInvokingAggregatorReturningMessageTests.java
View File

@@ -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;

36
spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests.java
View File

@@ -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));

64
spring-integration-core/src/test/java/org/springframework/integration/config/annotation/AggregatorAnnotationTests.java
View File

@@ -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);
}