From 753ce5635444a967444c841d0fbe9ef11a643d2d Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Wed, 5 Mar 2008 20:42:51 +0000 Subject: [PATCH] Added @Concurrency Annotation (INT-118). --- .../integration/annotation/Concurrency.java | 47 ++++++++++++++++ .../integration/bus/MessageBus.java | 42 +++++++++++++-- .../channel/DefaultChannelRegistry.java | 4 +- ...essageEndpointAnnotationPostProcessor.java | 15 +++--- .../dispatcher/DefaultMessageDispatcher.java | 9 ++++ .../dispatcher/MessageDispatcher.java | 2 + .../endpoint/ConcurrencyPolicy.java | 5 ++ .../endpoint/DefaultEndpointRegistry.java | 53 +++++++++++++++++++ .../endpoint/EndpointRegistry.java | 36 +++++++++++++ .../config/AggregatorAnnotationTests.java | 8 +-- ...eEndpointAnnotationPostProcessorTests.java | 24 +++++++++ 11 files changed, 226 insertions(+), 19 deletions(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/annotation/Concurrency.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/endpoint/DefaultEndpointRegistry.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/endpoint/EndpointRegistry.java diff --git a/spring-integration-core/src/main/java/org/springframework/integration/annotation/Concurrency.java b/spring-integration-core/src/main/java/org/springframework/integration/annotation/Concurrency.java new file mode 100644 index 0000000000..4178855098 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/annotation/Concurrency.java @@ -0,0 +1,47 @@ +/* + * 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.Inherited; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; + +import org.springframework.integration.endpoint.ConcurrencyPolicy; + +/** + * Defines the {@link ConcurrencyPolicy} settings for a {@link MessageEndpoint @MessageEndpoint}. + * + * @author Mark Fisher + */ +@Target(ElementType.TYPE) +@Retention(RetentionPolicy.RUNTIME) +@Inherited +@Documented +public @interface Concurrency { + + int coreSize() default ConcurrencyPolicy.DEFAULT_CORE_SIZE; + + int maxSize() default ConcurrencyPolicy.DEFAULT_MAX_SIZE; + + int queueCapacity() default ConcurrencyPolicy.DEFAULT_QUEUE_CAPACITY; + + int keepAliveSeconds() default ConcurrencyPolicy.DEFAULT_KEEP_ALIVE_SECONDS; + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java b/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java index 94029d0ed8..b9ca133849 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java @@ -16,8 +16,10 @@ package org.springframework.integration.bus; +import java.util.Collection; import java.util.List; import java.util.Map; +import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.ScheduledExecutorService; @@ -40,7 +42,9 @@ import org.springframework.integration.channel.SimpleChannel; import org.springframework.integration.dispatcher.DefaultMessageDispatcher; import org.springframework.integration.dispatcher.SchedulingMessageDispatcher; import org.springframework.integration.endpoint.ConcurrencyPolicy; +import org.springframework.integration.endpoint.DefaultEndpointRegistry; import org.springframework.integration.endpoint.DefaultMessageEndpoint; +import org.springframework.integration.endpoint.EndpointRegistry; import org.springframework.integration.endpoint.MessageEndpoint; import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.scheduling.MessagePublishingErrorHandler; @@ -57,7 +61,7 @@ import org.springframework.util.Assert; * * @author Mark Fisher */ -public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lifecycle { +public class MessageBus implements ChannelRegistry, EndpointRegistry, ApplicationContextAware, Lifecycle { public static final String ERROR_CHANNEL_NAME = "errorChannel"; @@ -68,7 +72,7 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif private final ChannelRegistry channelRegistry = new DefaultChannelRegistry(); - private final Map endpoints = new ConcurrentHashMap(); + private final EndpointRegistry endpointRegistry = new DefaultEndpointRegistry(); private final Map dispatchers = new ConcurrentHashMap(); @@ -253,7 +257,7 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif if (endpoint.getConcurrencyPolicy() == null && endpoint instanceof DefaultMessageEndpoint) { ((DefaultMessageEndpoint) endpoint).setConcurrencyPolicy(this.defaultConcurrencyPolicy); } - this.endpoints.put(name, endpoint); + this.endpointRegistry.registerEndpoint(name, endpoint); if (this.isRunning()) { activateEndpoint(endpoint); } @@ -262,9 +266,37 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif } } + public MessageEndpoint unregisterEndpoint(String name) { + MessageEndpoint endpoint = this.endpointRegistry.unregisterEndpoint(name); + if (endpoint == null) { + return null; + } + Collection dispatchers = this.dispatchers.values(); + boolean removed = false; + for (SchedulingMessageDispatcher dispatcher : dispatchers) { + removed = (removed || dispatcher.removeHandler(endpoint)); + } + if (removed) { + return endpoint; + } + return null; + } + + public MessageEndpoint lookupEndpoint(String endpointName) { + return this.endpointRegistry.lookupEndpoint(endpointName); + } + + public Set getEndpointNames() { + return this.endpointRegistry.getEndpointNames(); + } + private void activateEndpoints() { - for (MessageEndpoint endpoint : this.endpoints.values()) { - this.activateEndpoint(endpoint); + Set endpointNames = this.endpointRegistry.getEndpointNames(); + for (String name : endpointNames) { + MessageEndpoint endpoint = this.endpointRegistry.lookupEndpoint(name); + if (endpoint != null) { + this.activateEndpoint(endpoint); + } } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/DefaultChannelRegistry.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/DefaultChannelRegistry.java index 4e95c9b071..d531a3caed 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/DefaultChannelRegistry.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/DefaultChannelRegistry.java @@ -28,9 +28,9 @@ import org.springframework.util.Assert; */ public class DefaultChannelRegistry implements ChannelRegistry { - private Map channels = new ConcurrentHashMap(); + private final Map channels = new ConcurrentHashMap(); - private MessageChannel errorChannel; + private volatile MessageChannel errorChannel; public void setErrorChannel(MessageChannel errorChannel) { 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 fd2ddbead5..b0ea8ce2d0 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 @@ -42,6 +42,7 @@ 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.endpoint.ConcurrencyPolicy; import org.springframework.integration.endpoint.DefaultMessageEndpoint; import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.handler.MessageHandlerChain; @@ -103,16 +104,18 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor if (endpointAnnotation == null) { return bean; } - if (this.messageBus == null) { - if (logger.isWarnEnabled()) { - logger.warn(this.getClass().getSimpleName() + " is disabled since no 'messageBus' was provided"); - } - return bean; - } MessageHandlerChain handlerChain = this.createHandlerChain(bean); DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(handlerChain); this.configureInput(bean, beanName, endpointAnnotation, endpoint); this.configureDefaultOutput(bean, beanName, endpointAnnotation, endpoint); + Concurrency concurrencyAnnotation = bean.getClass().getAnnotation(Concurrency.class); + if (concurrencyAnnotation != null) { + ConcurrencyPolicy concurrencyPolicy = new ConcurrencyPolicy( + concurrencyAnnotation.coreSize(), concurrencyAnnotation.maxSize()); + concurrencyPolicy.setKeepAliveSeconds(concurrencyAnnotation.keepAliveSeconds()); + concurrencyPolicy.setQueueCapacity(concurrencyAnnotation.queueCapacity()); + endpoint.setConcurrencyPolicy(concurrencyPolicy); + } if (endpoint.getHandler() == null) { endpoint.setHandler(new MessageHandler() { public Message handle(Message message) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultMessageDispatcher.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultMessageDispatcher.java index 740606bc13..c5e9da4ce8 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultMessageDispatcher.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultMessageDispatcher.java @@ -111,6 +111,15 @@ public class DefaultMessageDispatcher implements SchedulingMessageDispatcher { } } + public boolean removeHandler(MessageHandler handler) { + boolean removed = false; + Collection> handlerLists = this.scheduledHandlers.values(); + for (List handlers : handlerLists) { + removed = (removed || handlers.remove(handler)); + } + return removed; + } + public boolean isRunning() { return this.running; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageDispatcher.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageDispatcher.java index 0d979fc358..7c2f4a37d9 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageDispatcher.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageDispatcher.java @@ -27,6 +27,8 @@ public interface MessageDispatcher { void addHandler(MessageHandler handler); + boolean removeHandler(MessageHandler handler); + int dispatch(); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ConcurrencyPolicy.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ConcurrencyPolicy.java index e4deeed585..bf5efb482f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ConcurrencyPolicy.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ConcurrencyPolicy.java @@ -89,4 +89,9 @@ public class ConcurrencyPolicy implements EndpointPolicy { this.keepAliveSeconds = keepAliveSeconds; } + public String toString() { + return "[coreSize=" + this.coreSize + ", maxSize=" + this.maxSize + + ", queueCapacity=" + this.queueCapacity + ", keepAliveSeconds=" + this.keepAliveSeconds + "]"; + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/DefaultEndpointRegistry.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/DefaultEndpointRegistry.java new file mode 100644 index 0000000000..797f52141a --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/DefaultEndpointRegistry.java @@ -0,0 +1,53 @@ +/* + * 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.endpoint; + +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; + +import org.springframework.util.Assert; + +/** + * A simple map-backed implementation of {@link EndpointRegistry}. + * + * @author Mark Fisher + */ +public class DefaultEndpointRegistry implements EndpointRegistry { + + private final Map endpoints = new ConcurrentHashMap(); + + + public MessageEndpoint lookupEndpoint(String endpointName) { + return this.endpoints.get(endpointName); + } + + public void registerEndpoint(String name, MessageEndpoint endpoint) { + Assert.notNull(name, "'name' must not be null"); + Assert.notNull(endpoint, "'endpoint' must not be null"); + this.endpoints.put(name, endpoint); + } + + public MessageEndpoint unregisterEndpoint(String name) { + return (name != null) ? this.endpoints.remove(name) : null; + } + + public Set getEndpointNames() { + return this.endpoints.keySet(); + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/EndpointRegistry.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/EndpointRegistry.java new file mode 100644 index 0000000000..081dcaba76 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/EndpointRegistry.java @@ -0,0 +1,36 @@ +/* + * 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.endpoint; + +import java.util.Set; + +/** + * A strategy interface for registration and lookup of message endpoints by name. + * + * @author Mark Fisher + */ +public interface EndpointRegistry { + + void registerEndpoint(String name, MessageEndpoint endpoint); + + MessageEndpoint unregisterEndpoint(String name); + + MessageEndpoint lookupEndpoint(String endpointName); + + Set getEndpointNames(); + +} 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 index 25a62ffeab..598ead965a 100644 --- 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 @@ -17,17 +17,16 @@ 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; @@ -85,10 +84,7 @@ public class AggregatorAnnotationTests { 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"); + DefaultMessageEndpoint endpoint = (DefaultMessageEndpoint) messageBus.lookupEndpoint(endpointName + "-endpoint"); ConcurrentHandler handler = (ConcurrentHandler) endpoint.getHandler(); DirectFieldAccessor concurrentHandlerAccessor = new DirectFieldAccessor(handler); MessageHandlerChain messageHandlerChain = (MessageHandlerChain) concurrentHandlerAccessor diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/annotation/MessageEndpointAnnotationPostProcessorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/annotation/MessageEndpointAnnotationPostProcessorTests.java index 64048d3072..5223585256 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/annotation/MessageEndpointAnnotationPostProcessorTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/annotation/MessageEndpointAnnotationPostProcessorTests.java @@ -25,6 +25,7 @@ import org.junit.Test; import org.springframework.context.support.AbstractApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; +import org.springframework.integration.annotation.Concurrency; import org.springframework.integration.annotation.DefaultOutput; import org.springframework.integration.annotation.MessageEndpoint; import org.springframework.integration.annotation.Polled; @@ -32,6 +33,8 @@ import org.springframework.integration.bus.MessageBus; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.SimpleChannel; import org.springframework.integration.config.MessageEndpointAnnotationPostProcessor; +import org.springframework.integration.endpoint.ConcurrencyPolicy; +import org.springframework.integration.endpoint.DefaultMessageEndpoint; import org.springframework.integration.message.Message; import org.springframework.integration.message.StringMessage; @@ -97,6 +100,21 @@ public class MessageEndpointAnnotationPostProcessorTests { messageBus.stop(); } + @Test + public void testConcurrencyAnnotationWithValues() { + MessageBus messageBus = new MessageBus(); + MessageEndpointAnnotationPostProcessor postProcessor = + new MessageEndpointAnnotationPostProcessor(messageBus); + ConcurrencyAnnotationTestBean testBean = new ConcurrencyAnnotationTestBean(); + postProcessor.postProcessAfterInitialization(testBean, "testBean"); + DefaultMessageEndpoint endpoint = (DefaultMessageEndpoint) messageBus.lookupEndpoint("testBean-endpoint"); + ConcurrencyPolicy concurrencyPolicy = endpoint.getConcurrencyPolicy(); + assertEquals(17, concurrencyPolicy.getCoreSize()); + assertEquals(42, concurrencyPolicy.getMaxSize()); + assertEquals(11, concurrencyPolicy.getQueueCapacity()); + assertEquals(123, concurrencyPolicy.getKeepAliveSeconds()); + } + @Test(expected=IllegalArgumentException.class) public void testPostProcessorWithNullMessageBus() { new MessageEndpointAnnotationPostProcessor(null); @@ -136,4 +154,10 @@ public class MessageEndpointAnnotationPostProcessorTests { } } + + @MessageEndpoint(input="inputChannel") + @Concurrency(coreSize=17, maxSize=42, keepAliveSeconds=123, queueCapacity=11) + private static class ConcurrencyAnnotationTestBean { + } + }