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

Using 'spring-integration-core-1.0.xsd' in 'testAnnotatedAggregator.xml' and added tests for wrong parameter counts in AggregatorAdapterTests.

Browse Source
This commit is contained in:
Mark Fisher
2008-03-04 16:03:29 +00:00
parent 2428a8390a
commit 0d1843211a
7 changed files with 158 additions and 94 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
spring-integration-core/src/main/java/org/springframework/integration/annotation/Aggregator.java
View File

@@ -34,11 +34,19 @@ import org.springframework.integration.router.AggregatingMessageHandler;
@Documented
@Handler
public @interface Aggregator {
String defaultReplyChannel() default "";
String discardChannel() default "";
long sendTimeout() default AggregatingMessageHandler.DEFAULT_SEND_TIMEOUT;
long timeout() default AggregatingMessageHandler.DEFAULT_TIMEOUT;
boolean sendPartialResultsOnTimeout() default false;
long reaperInterval() default AggregatingMessageHandler.DEFAULT_REAPER_INTERVAL;
int trackedCorrelationIdCapacity() default AggregatingMessageHandler.DEFAULT_TRACKED_CORRRELATION_ID_CAPACITY;
String defaultReplyChannel() default "";
String discardChannel() default "";
long sendTimeout() default AggregatingMessageHandler.DEFAULT_SEND_TIMEOUT;
long timeout() default AggregatingMessageHandler.DEFAULT_TIMEOUT;
boolean sendPartialResultsOnTimeout() default false;
long reaperInterval() default AggregatingMessageHandler.DEFAULT_REAPER_INTERVAL;
int trackedCorrelationIdCapacity() default AggregatingMessageHandler.DEFAULT_TRACKED_CORRRELATION_ID_CAPACITY;
}

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

@@ -76,13 +76,13 @@ public class AggregatorParser implements BeanDefinitionParser {
return parseAggregatorElement(element, parserContext, true);
}
public static BeanDefinition parseAggregatorElement(Element element, ParserContext parserContext, boolean topLevel) {
private BeanDefinition parseAggregatorElement(Element element, ParserContext parserContext, boolean topLevel) {
final RootBeanDefinition aggregatorDef = new RootBeanDefinition(AggregatingMessageHandler.class);
aggregatorDef.setSource(parserContext.extractSource(element));
final String id = element.getAttribute(ID_ATTRIBUTE);
final String ref = element.getAttribute(REF_ATTRIBUTE);
final String method = element.getAttribute(METHOD_ATTRIBUTE);
if (!StringUtils.hasText(ref)) {
throw new MessagingConfigurationException("The 'ref' attribute must be present");
}
@@ -101,7 +101,6 @@ public class AggregatorParser implements BeanDefinitionParser {
String adapterBeanName = parserContext.getReaderContext().generateBeanName(adapterDefinition);
parserContext.registerBeanComponent(new BeanComponentDefinition(adapterDefinition, adapterBeanName));
aggregatorDef.getConstructorArgumentValues().addGenericArgumentValue(new RuntimeBeanReference(adapterBeanName));
}
IntegrationNamespaceUtils.setBeanReferenceIfAttributeDefined(aggregatorDef, COMPLETION_STRATEGY_PROPERTY,
element, COMPLETION_STRATEGY_ATTRIBUTE);

9
spring-integration-core/src/main/java/org/springframework/integration/router/AggregatingMessageHandler.java
View File

@@ -63,13 +63,14 @@ import org.springframework.util.CollectionUtils;
public class AggregatingMessageHandler implements MessageHandler, InitializingBean {
public final static long DEFAULT_SEND_TIMEOUT = 1000;
public final static long DEFAULT_TIMEOUT = 60000;
public final static long DEFAULT_REAPER_INTERVAL = 1000;
public final static int DEFAULT_TRACKED_CORRRELATION_ID_CAPACITY = 1000;
private final Log logger = LogFactory.getLog(this.getClass());
private final Aggregator aggregator;

56
spring-integration-core/src/main/java/org/springframework/integration/router/AggregatorAdapter.java
View File

@@ -21,8 +21,8 @@ import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import org.springframework.integration.MessagingConfigurationException;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.Message;
import org.springframework.integration.util.SimpleMethodInvoker;
@@ -30,10 +30,12 @@ import org.springframework.util.Assert;
import org.springframework.util.ReflectionUtils;
/**
* Aggregator adapter for methods annotated with {@link org.springframework.integration.annotation.Aggregator @Aggregator} and for
* <aggregator ref="beanReference"method="methodName"/>
* Aggregator adapter for methods annotated with {@link org.springframework.integration.annotation.Aggregator @Aggregator}
* and for '<code>aggregator</code>' elements that include a '<code>method</code>' attribute
* (e.g. &lt;aggregator ref="beanReference" method="methodName"/&gt;).
*
* @author Marius Bogoevici
* @author Mark Fisher
*/
public class AggregatorAdapter implements Aggregator {
@@ -41,46 +43,52 @@ public class AggregatorAdapter implements Aggregator {
private final Method method;
public AggregatorAdapter(Object object, String methodName) {
Assert.notNull(object, "'object' must not be null");
Assert.notNull(methodName, "'methodName' must not be null.");
Assert.notNull(methodName, "'methodName' must not be null");
this.method = ReflectionUtils.findMethod(object.getClass(), methodName, new Class<?>[] { Collection.class });
Assert
.notNull(this.method, "Method '" + methodName + "'(Collection<?> args) not found on "
+ object.getClass());
this.invoker = new SimpleMethodInvoker<Object>(object, method.getName());
Assert.notNull(this.method, "Method '" + methodName + "(Collection<?> args)' not found on '" +
object.getClass().getName() + "'.");
this.invoker = new SimpleMethodInvoker<Object>(object, this.method.getName());
}
public AggregatorAdapter(Object object, Method method) {
Assert.notNull(object, "'object' must not be null");
Assert.notNull(method, "'method' must not be null.");
if (method.getParameterTypes().length != 1 && method.getParameterTypes()[0].equals(Collection.class)) {
throw new MessagingConfigurationException(
Assert.notNull(method, "'method' must not be null");
if (method.getParameterTypes().length != 1 || !method.getParameterTypes()[0].equals(Collection.class)) {
throw new IllegalArgumentException(
"Aggregator method must accept exactly one parameter, and it must be a Collection.");
}
this.method = method;
this.invoker = new SimpleMethodInvoker<Object>(object, method.getName());
this.invoker = new SimpleMethodInvoker<Object>(object, this.method.getName());
}
public Message<?> aggregate(Collection<Message<?>> messages) {
Object returnedValue = null;
if (isMethodParameterParametrized(method) && isHavingActualTypeArguments(method)
&& (isActualTypeRawMessage(method) || isActualTypeParametrizedMessage(method))) {
returnedValue = invoker.invokeMethod(messages);
if (isMethodParameterParametrized(this.method) && isHavingActualTypeArguments(this.method)
&& (isActualTypeRawMessage(this.method) || isActualTypeParametrizedMessage(this.method))) {
returnedValue = this.invoker.invokeMethod(messages);
}
else {
returnedValue = invoker.invokeMethod(extractPayloadsFromMessages(messages));
returnedValue = this.invoker.invokeMethod(extractPayloadsFromMessages(messages));
}
if (returnedValue == null) {
return null;
}
else if (returnedValue instanceof Message) {
if (returnedValue instanceof Message) {
return (Message<?>) returnedValue;
}
else {
return new GenericMessage<Object>(returnedValue);
return new GenericMessage<Object>(returnedValue);
}
private Collection<?> extractPayloadsFromMessages(Collection<Message<?>> messages) {
List<Object> payloadList = new ArrayList<Object>();
for (Message<?> message : messages) {
payloadList.add(message.getPayload());
}
return payloadList;
}
private static boolean isActualTypeParametrizedMessage(Method method) {
@@ -106,12 +114,4 @@ public class AggregatorAdapter implements Aggregator {
&& method.getGenericParameterTypes()[0] instanceof ParameterizedType;
}
private Collection<?> extractPayloadsFromMessages(Collection<Message<?>> messages) {
ArrayList payloadList = new ArrayList<Object>();
for (Message<?> message : messages) {
payloadList.add(message.getPayload());
}
return payloadList;
}
}

55
spring-integration-core/src/main/java/org/springframework/integration/router/config/AggregatorMessageHandlerCreator.java
View File

@@ -26,26 +26,27 @@ import org.springframework.integration.router.AggregatingMessageHandler;
import org.springframework.integration.router.AggregatorAdapter;
/**
* Creates a {@link AggregatorAdapter AggregatorAdapter} adapter for methods that aggregate messages.
* Creates an {@link AggregatorAdapter AggregatorAdapter} for methods that aggregate messages.
*
* @author Marius Bogoevici
*/
public class AggregatorMessageHandlerCreator extends AbstractMessageHandlerCreator {
private static final String DEFAULT_REPLY_CHANNEL = "defaultReplyChannel";
private static final String DISCARD_CHANNEL = "discardChannel";
private static final String SEND_TIMEOUT = "sendTimeout";
private static final String SEND_PARTIAL_RESULTS_ON_TIMEOUT = "sendPartialResultsOnTimeout";
private static final String REAPER_INTERVAL = "reaperInterval";
private static final String TIMEOUT = "timeout";
private static final String TRACKED_CORRELATION_ID_CAPACITY = "trackedCorrelationIdCapacity";
private final MessageBus messageBus;
@@ -53,23 +54,31 @@ public class AggregatorMessageHandlerCreator extends AbstractMessageHandlerCreat
this.messageBus = messageBus;
}
public MessageHandler doCreateHandler(Object object, Method method, Map<String, ?> attributes) {
AggregatingMessageHandler messageHandler = new AggregatingMessageHandler(new AggregatorAdapter(object, method));
if (attributes.containsKey(DEFAULT_REPLY_CHANNEL))
messageHandler.setDefaultReplyChannel(messageBus.lookupChannel((String)attributes.get(DEFAULT_REPLY_CHANNEL)));
if (attributes.containsKey(DISCARD_CHANNEL))
messageHandler.setDiscardChannel(messageBus.lookupChannel((String)attributes.get(DISCARD_CHANNEL)));
if (attributes.containsKey(SEND_TIMEOUT))
messageHandler.setSendTimeout((Long)attributes.get(SEND_TIMEOUT));
if (attributes.containsKey(SEND_PARTIAL_RESULTS_ON_TIMEOUT))
messageHandler.setSendPartialResultOnTimeout((Boolean)attributes.get(SEND_PARTIAL_RESULTS_ON_TIMEOUT));
if (attributes.containsKey(REAPER_INTERVAL))
messageHandler.setReaperInterval((Long)attributes.get(REAPER_INTERVAL));
if(attributes.containsKey(TIMEOUT))
messageHandler.setTimeout((Long)attributes.get(TIMEOUT));
if(attributes.containsKey(TRACKED_CORRELATION_ID_CAPACITY))
messageHandler.setTrackedCorrelationIdCapacity((Integer)attributes.get(TRACKED_CORRELATION_ID_CAPACITY));
if (attributes.containsKey(DEFAULT_REPLY_CHANNEL)) {
messageHandler.setDefaultReplyChannel(this.messageBus.lookupChannel((String) attributes.get(DEFAULT_REPLY_CHANNEL)));
}
if (attributes.containsKey(DISCARD_CHANNEL)) {
messageHandler.setDiscardChannel(this.messageBus.lookupChannel((String) attributes.get(DISCARD_CHANNEL)));
}
if (attributes.containsKey(SEND_TIMEOUT)) {
messageHandler.setSendTimeout((Long) attributes.get(SEND_TIMEOUT));
}
if (attributes.containsKey(SEND_PARTIAL_RESULTS_ON_TIMEOUT)) {
messageHandler.setSendPartialResultOnTimeout((Boolean) attributes.get(SEND_PARTIAL_RESULTS_ON_TIMEOUT));
}
if (attributes.containsKey(REAPER_INTERVAL)) {
messageHandler.setReaperInterval((Long) attributes.get(REAPER_INTERVAL));
}
if (attributes.containsKey(TIMEOUT)) {
messageHandler.setTimeout((Long) attributes.get(TIMEOUT));
}
if (attributes.containsKey(TRACKED_CORRELATION_ID_CAPACITY)) {
messageHandler.setTrackedCorrelationIdCapacity((Integer) attributes.get(TRACKED_CORRELATION_ID_CAPACITY));
}
return messageHandler;
}
}
}

27
spring-integration-core/src/test/java/org/springframework/integration/config/testAnnotatedAggregator.xml
View File

@@ -5,22 +5,23 @@
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration-1.0.xsd
http://www.springframework.org/schema/integration/spring-integration-core-1.0.xsd
http://www.springframework.org/schema/context
http://www.springframework.org/schema/context/spring-context-2.5.xsd">
<message-bus />
<annotation-driven />
<message-bus/>
<annotation-driven/>
<channel id="inputChannel"/>
<channel id="replyChannel"/>
<channel id="discardChannel"/>
<context:component-scan base-package="org.springframework.integration.config" use-default-filters="false">
<context:include-filter type="regex"
expression="org\.springframework\.integration\.config\.TestAnnotatedEndpointWithDefaultAggregator" />
<context:include-filter type="regex"
expression="org\.springframework\.integration\.config\.TestAnnotatedEndpointWithCustomizedAggregator" />
expression="org\.springframework\.integration\.config\.TestAnnotatedEndpoint.*"/>
</context:component-scan>
<channel id="inputChannel" />
<channel id="replyChannel" />
<channel id="discardChannel" />
</beans:beans>
</beans:beans>

78
spring-integration-core/src/test/java/org/springframework/integration/router/AggregatorAdapterTests.java
View File

@@ -17,24 +17,24 @@
package org.springframework.integration.router;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.springframework.integration.MessagingConfigurationException;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.Message;
/**
* @author Marius Bogoevici
* @author Mark Fisher
*/
public class AggregatorAdapterTests {
private SimpleAggregator simpleAggregator;
@Before
public void setUp() {
simpleAggregator = new SimpleAggregator();
@@ -87,9 +87,43 @@ public class AggregatorAdapterTests {
@Test(expected=IllegalArgumentException.class)
public void testAdapterWithWrongMethodName() {
Aggregator aggregator = new AggregatorAdapter(simpleAggregator, "methodThatDoesNotExist");
new AggregatorAdapter(simpleAggregator, "methodThatDoesNotExist");
}
@Test(expected=IllegalArgumentException.class)
public void testInvalidParameterTypeUsingMethodName() {
new AggregatorAdapter(simpleAggregator, "invalidParameterType");
}
@Test(expected=IllegalArgumentException.class)
public void testTooManyParametersUsingMethodName() {
new AggregatorAdapter(simpleAggregator, "tooManyParameters");
}
@Test(expected=IllegalArgumentException.class)
public void testNotEnoughParametersUsingMethodName() {
new AggregatorAdapter(simpleAggregator, "notEnoughParameters");
}
@Test(expected=IllegalArgumentException.class)
public void testInvalidParameterTypeUsingMethodObject() throws SecurityException, NoSuchMethodException {
new AggregatorAdapter(simpleAggregator, simpleAggregator.getClass().getMethod(
"invalidParameterType", String.class));
}
@Test(expected=IllegalArgumentException.class)
public void testTooManyParametersUsingMethodObject() throws SecurityException, NoSuchMethodException {
new AggregatorAdapter(simpleAggregator, simpleAggregator.getClass().getMethod(
"tooManyParameters", Collection.class, Collection.class));
}
@Test(expected=IllegalArgumentException.class)
public void testNotEnoughParametersUsingMethodObject() throws SecurityException, NoSuchMethodException {
new AggregatorAdapter(simpleAggregator, simpleAggregator.getClass().getMethod(
"notEnoughParameters", null));
}
private static Collection<Message<?>> createCollectionOfMessages() {
Collection<Message<?>> messages = new ArrayList<Message<?>>();
messages.add(new GenericMessage<String>("123"));
@@ -98,17 +132,22 @@ public class AggregatorAdapterTests {
return messages;
}
private class SimpleAggregator {
private volatile boolean aggregationPerformed;
public SimpleAggregator() {
aggregationPerformed = false;
this.aggregationPerformed = false;
}
public boolean isAggregationPerformed() {
return this.aggregationPerformed;
}
public Message<?> doAggregationOnCollectionOfMessages(Collection<Message> messages) {
aggregationPerformed = true;
this.aggregationPerformed = true;
StringBuffer buffer = new StringBuffer();
for (Message<?> message : messages) {
buffer.append(message.getPayload());
@@ -117,16 +156,16 @@ public class AggregatorAdapterTests {
}
public Message<?> doAggregationOnCollectionOfMessagesParametrizedWithWildcard(Collection<Message<?>> messages) {
aggregationPerformed = true;
this.aggregationPerformed = true;
StringBuffer buffer = new StringBuffer();
for (Message<?> message : messages) {
buffer.append(message.getPayload());
}
return new GenericMessage<String>(buffer.toString());
}
public Message<?> doAggregationOnCollectionOfMessagesParametrizedWithString(Collection<Message<String>> messages) {
aggregationPerformed = true;
this.aggregationPerformed = true;
StringBuffer buffer = new StringBuffer();
for (Message<String> message : messages) {
buffer.append(message.getPayload());
@@ -135,7 +174,7 @@ public class AggregatorAdapterTests {
}
public Message<?> doAggregationOnCollectionOfStrings(Collection<String> messages) {
aggregationPerformed = true;
this.aggregationPerformed = true;
StringBuffer buffer = new StringBuffer();
for (String payload : messages) {
buffer.append(payload);
@@ -144,7 +183,7 @@ public class AggregatorAdapterTests {
}
public Long doAggregationOnCollectionOfStringsReturningLong(Collection<String> messages) {
aggregationPerformed = true;
this.aggregationPerformed = true;
StringBuffer buffer = new StringBuffer();
for (String payload : messages) {
buffer.append(payload);
@@ -152,10 +191,17 @@ public class AggregatorAdapterTests {
return Long.parseLong(buffer.toString());
}
public boolean isAggregationPerformed() {
return aggregationPerformed;
public Message<?> invalidParameterType(String invalid) {
return null;
}
public Message<?> tooManyParameters(Collection<?> c1, Collection<?> c2) {
return null;
}
public Message<?> notEnoughParameters() {
return null;
}
}
}