diff --git a/spring-integration-core/src/main/java/org/springframework/integration/annotation/Aggregator.java b/spring-integration-core/src/main/java/org/springframework/integration/annotation/Aggregator.java new file mode 100644 index 0000000000..536e8eec8d --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/annotation/Aggregator.java @@ -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; +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java index a467a03ac4..93983eaefc 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java @@ -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 { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/router/AggregatingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/router/AggregatingMessageHandler.java index b042a72835..ed51eb6641 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/router/AggregatingMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/router/AggregatingMessageHandler.java @@ -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 barriers = new ConcurrentHashMap(); - 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 trackedCorrelationIds; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/router/AggregatorAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/router/AggregatorAdapter.java index b08f81292c..6b2bc3fc65 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/router/AggregatorAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/router/AggregatorAdapter.java @@ -30,8 +30,8 @@ import org.springframework.util.Assert; import org.springframework.util.ReflectionUtils; /** - * Aggregator adapter for methods used through + * Aggregator adapter for methods annotated with {@link org.springframework.integration.annotation.Aggregator @Aggregator} and for + * <aggregator ref="beanReference"method="methodName"/> * * @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, 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, method.getName()); } public Message aggregate(Collection> messages) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/router/config/AggregatorMessageHandlerCreator.java b/spring-integration-core/src/main/java/org/springframework/integration/router/config/AggregatorMessageHandlerCreator.java new file mode 100644 index 0000000000..27da5beda7 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/router/config/AggregatorMessageHandlerCreator.java @@ -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 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; + } + +} \ No newline at end of file diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorAnnotationTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorAnnotationTests.java new file mode 100644 index 0000000000..25a62ffeab --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorAnnotationTests.java @@ -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 endpoints = (Map) 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; + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/TestAnnotatedEndpointWithCustomizedAggregator.java b/spring-integration-core/src/test/java/org/springframework/integration/config/TestAnnotatedEndpointWithCustomizedAggregator.java new file mode 100644 index 0000000000..039600f43a --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/TestAnnotatedEndpointWithCustomizedAggregator.java @@ -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> aggregatedMessages = new ConcurrentHashMap>(); + + @Aggregator(defaultReplyChannel = "replyChannel", discardChannel = "discardChannel", + reaperInterval = 1234, sendPartialResultsOnTimeout = true, + sendTimeout = 98765432, timeout = 4567890, trackedCorrelationIdCapacity = 42) + public Message aggregatingMethod(Collection> messages) { + List> sortableList = new ArrayList>(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> getAggregatedMessages() { + return aggregatedMessages; + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/TestAnnotatedEndpointWithDefaultAggregator.java b/spring-integration-core/src/test/java/org/springframework/integration/config/TestAnnotatedEndpointWithDefaultAggregator.java new file mode 100644 index 0000000000..8db15e2d63 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/TestAnnotatedEndpointWithDefaultAggregator.java @@ -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> aggregatedMessages = new ConcurrentHashMap>(); + + @Aggregator + public Message aggregatingMethod(Collection> messages) { + List> sortableList = new ArrayList>(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> getAggregatedMessages() { + return aggregatedMessages; + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/testAnnotatedAggregator.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/testAnnotatedAggregator.xml new file mode 100644 index 0000000000..4831d2bf3b --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/testAnnotatedAggregator.xml @@ -0,0 +1,26 @@ + + + + + + + + + + + + + + + + \ No newline at end of file