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

Add @Aggregator annotation for annotating methods that can aggregate messages (INT-94).

Browse Source
This commit is contained in:
Marius Bogoevici
2008-03-04 05:23:06 +00:00
parent 404c5f2727
commit 2428a8390a
9 changed files with 401 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

44
spring-integration-core/src/main/java/org/springframework/integration/annotation/Aggregator.java Normal file
View File

@@ -0,0 +1,44 @@
/*
* Copyright 2002-2007 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.annotation;
import java.lang.annotation.Documented;
import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
import org.springframework.integration.router.AggregatingMessageHandler;
/**
* Indicates that a method is capable of aggregating messages.
*
* @author Marius Bogoevici
*/
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
@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;
}

10
spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java
View File

@@ -37,16 +37,12 @@ import org.springframework.integration.adapter.DefaultTargetAdapter;
import org.springframework.integration.adapter.MethodInvokingSource;
import org.springframework.integration.adapter.MethodInvokingTarget;
import org.springframework.integration.adapter.PollingSourceAdapter;
import org.springframework.integration.annotation.DefaultOutput;
import org.springframework.integration.annotation.Handler;
import org.springframework.integration.annotation.MessageEndpoint;
import org.springframework.integration.annotation.Polled;
import org.springframework.integration.annotation.Router;
import org.springframework.integration.annotation.Splitter;
import org.springframework.integration.annotation.*;
import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.channel.ChannelRegistryAware;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.SimpleChannel;
import org.springframework.integration.channel.ChannelRegistry;
import org.springframework.integration.endpoint.DefaultMessageEndpoint;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.handler.MessageHandlerChain;
@@ -55,6 +51,7 @@ import org.springframework.integration.handler.config.MessageHandlerCreator;
import org.springframework.integration.message.Message;
import org.springframework.integration.router.config.RouterMessageHandlerCreator;
import org.springframework.integration.router.config.SplitterMessageHandlerCreator;
import org.springframework.integration.router.config.AggregatorMessageHandlerCreator;
import org.springframework.integration.scheduling.PollingSchedule;
import org.springframework.integration.scheduling.Schedule;
import org.springframework.integration.scheduling.Subscription;
@@ -95,6 +92,7 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor
this.handlerCreators.put(Handler.class, new DefaultMessageHandlerCreator());
this.handlerCreators.put(Router.class, new RouterMessageHandlerCreator());
this.handlerCreators.put(Splitter.class, new SplitterMessageHandlerCreator());
this.handlerCreators.put(Aggregator.class, new AggregatorMessageHandlerCreator(messageBus));
}
public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {

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

@@ -62,6 +62,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;
@@ -70,19 +78,19 @@ public class AggregatingMessageHandler implements MessageHandler, InitializingBe
private volatile MessageChannel discardChannel;
private volatile long sendTimeout = 1000;
private volatile long sendTimeout = DEFAULT_SEND_TIMEOUT;
private volatile CompletionStrategy completionStrategy = new SequenceSizeCompletionStrategy();
private final ConcurrentMap<Object, AggregationBarrier> barriers = new ConcurrentHashMap<Object, AggregationBarrier>();
private volatile long timeout = 60000;
private volatile long timeout = DEFAULT_TIMEOUT;
private volatile boolean sendPartialResultOnTimeout = false;
private volatile long reaperInterval = 1000;
private volatile long reaperInterval = DEFAULT_REAPER_INTERVAL;
private volatile int trackedCorrelationIdCapacity = 1000;
private volatile int trackedCorrelationIdCapacity = DEFAULT_TRACKED_CORRRELATION_ID_CAPACITY;
private volatile BlockingQueue<Object> trackedCorrelationIds;

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

@@ -30,8 +30,8 @@ import org.springframework.util.Assert;
import org.springframework.util.ReflectionUtils;
/**
* Aggregator adapter for methods used through <aggregator
* ref="beanReference"method="methodName"/>
* Aggregator adapter for methods annotated with {@link org.springframework.integration.annotation.Aggregator @Aggregator} and for
* &lt;aggregator ref="beanReference"method="methodName"/&gt;
*
* @author Marius Bogoevici
*/
@@ -54,16 +54,12 @@ public class AggregatorAdapter implements Aggregator {
public AggregatorAdapter(Object object, Method method) {
Assert.notNull(object, "'object' must not be null");
Assert.notNull(method, "'method' must not be null.");
assertMethodCanBeAdapted();
this.method = method;
this.invoker = new SimpleMethodInvoker<Object>(object, method.getName());
}
private void assertMethodCanBeAdapted() {
if (this.method.getParameterTypes().length != 1 && this.method.getParameterTypes()[0].equals(Collection.class)) {
if (method.getParameterTypes().length != 1 && method.getParameterTypes()[0].equals(Collection.class)) {
throw new MessagingConfigurationException(
"Aggregator method must accept exactly one parameter, and it must be a Collection.");
}
this.method = method;
this.invoker = new SimpleMethodInvoker<Object>(object, method.getName());
}
public Message<?> aggregate(Collection<Message<?>> messages) {

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

@@ -0,0 +1,75 @@
/*
* Copyright 2002-2007 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.router.config;
import java.lang.reflect.Method;
import java.util.Map;
import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.handler.config.AbstractMessageHandlerCreator;
import org.springframework.integration.router.AggregatingMessageHandler;
import org.springframework.integration.router.AggregatorAdapter;
/**
* Creates a {@link AggregatorAdapter AggregatorAdapter} adapter 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;
public AggregatorMessageHandlerCreator(MessageBus messageBus) {
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));
return messageHandler;
}
}

107
spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorAnnotationTests.java Normal file
View File

@@ -0,0 +1,107 @@
/*
* Copyright 2002-2007 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.util.List;
import java.util.Map;
import org.junit.Assert;
import org.junit.Test;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.endpoint.ConcurrentHandler;
import org.springframework.integration.endpoint.DefaultMessageEndpoint;
import org.springframework.integration.endpoint.MessageEndpoint;
import org.springframework.integration.handler.MessageHandlerChain;
import org.springframework.integration.router.AggregatingMessageHandler;
import org.springframework.integration.router.SequenceSizeCompletionStrategy;
/**
* @author Marius Bogoevici
*/
public class AggregatorAnnotationTests {
@Test
public void testAnnotationWithDefaultSettings() {
ApplicationContext context = new ClassPathXmlApplicationContext(
new String[] { "classpath:/org/springframework/integration/config/testAnnotatedAggregator.xml" });
final String endpointName = "endpointWithDefaultAnnotation";
DirectFieldAccessor aggregatingMessageHandlerAccessor = getDirectFieldAccessorForAggregatingHandler(context,
endpointName);
Assert.assertTrue(aggregatingMessageHandlerAccessor.getPropertyValue("completionStrategy") instanceof SequenceSizeCompletionStrategy);
Assert.assertNull(aggregatingMessageHandlerAccessor.getPropertyValue("defaultReplyChannel"));
Assert.assertNull(aggregatingMessageHandlerAccessor.getPropertyValue("discardChannel"));
Assert.assertEquals(AggregatingMessageHandler.DEFAULT_SEND_TIMEOUT, aggregatingMessageHandlerAccessor
.getPropertyValue("sendTimeout"));
Assert.assertEquals(AggregatingMessageHandler.DEFAULT_TIMEOUT, aggregatingMessageHandlerAccessor
.getPropertyValue("timeout"));
Assert.assertEquals(false, aggregatingMessageHandlerAccessor.getPropertyValue("sendPartialResultOnTimeout"));
Assert.assertEquals(AggregatingMessageHandler.DEFAULT_REAPER_INTERVAL, aggregatingMessageHandlerAccessor
.getPropertyValue("reaperInterval"));
Assert.assertEquals(AggregatingMessageHandler.DEFAULT_TRACKED_CORRRELATION_ID_CAPACITY,
aggregatingMessageHandlerAccessor.getPropertyValue("trackedCorrelationIdCapacity"));
}
@Test
public void testAnnotationWithCustomSettings() {
ApplicationContext context = new ClassPathXmlApplicationContext(
new String[] { "classpath:/org/springframework/integration/config/testAnnotatedAggregator.xml" });
final String endpointName = "endpointWithCustomizedAnnotation";
DirectFieldAccessor aggregatingMessageHandlerAccessor = getDirectFieldAccessorForAggregatingHandler(context,
endpointName);
Assert.assertTrue(aggregatingMessageHandlerAccessor.getPropertyValue("completionStrategy") instanceof SequenceSizeCompletionStrategy);
Assert.assertEquals(getMessageBus(context).lookupChannel("replyChannel"), aggregatingMessageHandlerAccessor
.getPropertyValue("defaultReplyChannel"));
Assert.assertEquals(getMessageBus(context).lookupChannel("discardChannel"), aggregatingMessageHandlerAccessor
.getPropertyValue("discardChannel"));
Assert.assertEquals(98765432l, aggregatingMessageHandlerAccessor
.getPropertyValue("sendTimeout"));
Assert.assertEquals(4567890l, aggregatingMessageHandlerAccessor
.getPropertyValue("timeout"));
Assert.assertEquals(true, aggregatingMessageHandlerAccessor.getPropertyValue("sendPartialResultOnTimeout"));
Assert.assertEquals(1234l, aggregatingMessageHandlerAccessor
.getPropertyValue("reaperInterval"));
Assert.assertEquals(42,
aggregatingMessageHandlerAccessor.getPropertyValue("trackedCorrelationIdCapacity"));
}
@SuppressWarnings("unchecked")
private DirectFieldAccessor getDirectFieldAccessorForAggregatingHandler(ApplicationContext context,
final String endpointName) {
MessageBus messageBus = getMessageBus(context);
DirectFieldAccessor messageBusAccessor = new DirectFieldAccessor(messageBus);
Map<String, MessageEndpoint> endpoints = (Map<String, MessageEndpoint>) messageBusAccessor
.getPropertyValue("endpoints");
DefaultMessageEndpoint endpoint = (DefaultMessageEndpoint) endpoints.get(endpointName + "-endpoint");
ConcurrentHandler handler = (ConcurrentHandler) endpoint.getHandler();
DirectFieldAccessor concurrentHandlerAccessor = new DirectFieldAccessor(handler);
MessageHandlerChain messageHandlerChain = (MessageHandlerChain) concurrentHandlerAccessor
.getPropertyValue("handler");
AggregatingMessageHandler aggregatingMessageHandler = (AggregatingMessageHandler) ((List) new DirectFieldAccessor(
messageHandlerChain).getPropertyValue("handlers")).get(0);
DirectFieldAccessor aggregatingMessageHandlerAccessor = new DirectFieldAccessor(aggregatingMessageHandler);
return aggregatingMessageHandlerAccessor;
}
private MessageBus getMessageBus(ApplicationContext context) {
MessageBus messageBus = (MessageBus) context.getBean(MessageBusParser.MESSAGE_BUS_BEAN_NAME);
return messageBus;
}
}

65
spring-integration-core/src/test/java/org/springframework/integration/config/TestAnnotatedEndpointWithCustomizedAggregator.java Normal file
View File

@@ -0,0 +1,65 @@
/*
* Copyright 2002-2007 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.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import org.springframework.integration.annotation.Aggregator;
import org.springframework.integration.annotation.MessageEndpoint;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.StringMessage;
import org.springframework.integration.router.MessageSequenceComparator;
import org.springframework.stereotype.Component;
/**
* @author Marius Bogoevici
*/
@MessageEndpoint(input = "inputChannel")
@Component("endpointWithCustomizedAnnotation")
public class TestAnnotatedEndpointWithCustomizedAggregator {
private final ConcurrentMap<Object, Message<?>> aggregatedMessages = new ConcurrentHashMap<Object, Message<?>>();
@Aggregator(defaultReplyChannel = "replyChannel", discardChannel = "discardChannel",
reaperInterval = 1234, sendPartialResultsOnTimeout = true,
sendTimeout = 98765432, timeout = 4567890, trackedCorrelationIdCapacity = 42)
public Message<?> aggregatingMethod(Collection<Message<?>> messages) {
List<Message<?>> sortableList = new ArrayList<Message<?>>(messages);
Collections.sort(sortableList, new MessageSequenceComparator());
StringBuffer buffer = new StringBuffer();
Object correlationId = null;
for (Message<?> message : sortableList) {
buffer.append(message.getPayload().toString());
if (null == correlationId) {
correlationId = message.getHeader().getCorrelationId();
}
}
Message<?> returnedMessage = new StringMessage(buffer.toString());
aggregatedMessages.put(correlationId, returnedMessage);
return returnedMessage;
}
public ConcurrentMap<Object, Message<?>> getAggregatedMessages() {
return aggregatedMessages;
}
}

63
spring-integration-core/src/test/java/org/springframework/integration/config/TestAnnotatedEndpointWithDefaultAggregator.java Normal file
View File

@@ -0,0 +1,63 @@
/*
* Copyright 2002-2007 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.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import org.springframework.integration.annotation.Aggregator;
import org.springframework.integration.annotation.MessageEndpoint;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.StringMessage;
import org.springframework.integration.router.MessageSequenceComparator;
import org.springframework.stereotype.Component;
/**
* @author Marius Bogoevici
*/
@MessageEndpoint(input="inputChannel")
@Component("endpointWithDefaultAnnotation")
public class TestAnnotatedEndpointWithDefaultAggregator {
private final ConcurrentMap<Object, Message<?>> aggregatedMessages = new ConcurrentHashMap<Object, Message<?>>();
@Aggregator
public Message<?> aggregatingMethod(Collection<Message<?>> messages) {
List<Message<?>> sortableList = new ArrayList<Message<?>>(messages);
Collections.sort(sortableList, new MessageSequenceComparator());
StringBuffer buffer = new StringBuffer();
Object correlationId = null;
for (Message<?> message : sortableList) {
buffer.append(message.getPayload().toString());
if (null == correlationId) {
correlationId = message.getHeader().getCorrelationId();
}
}
Message<?> returnedMessage = new StringMessage(buffer.toString());
aggregatedMessages.put(correlationId, returnedMessage);
return returnedMessage;
}
public ConcurrentMap<Object, Message<?>> getAggregatedMessages() {
return aggregatedMessages;
}
}

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

@@ -0,0 +1,26 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns="http://www.springframework.org/schema/integration"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:beans="http://www.springframework.org/schema/beans"
xmlns:aop="http://www.springframework.org/schema/aop" xmlns:context="http://www.springframework.org/schema/context"
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/context
http://www.springframework.org/schema/context/spring-context-2.5.xsd">
<message-bus />
<annotation-driven />
<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" />
</context:component-scan>
<channel id="inputChannel" />
<channel id="replyChannel" />
<channel id="discardChannel" />
</beans:beans>