Added @Concurrency Annotation (INT-118).

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;
+
+}

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<String, MessageEndpoint> endpoints = new ConcurrentHashMap<String, MessageEndpoint>();
+	private final EndpointRegistry endpointRegistry = new DefaultEndpointRegistry();
 
 	private final Map<MessageChannel, SchedulingMessageDispatcher> dispatchers = new ConcurrentHashMap<MessageChannel, SchedulingMessageDispatcher>();
 
@@ -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<SchedulingMessageDispatcher> 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<String> getEndpointNames() {
+		return this.endpointRegistry.getEndpointNames();
+	}
+
 	private void activateEndpoints() {
-		for (MessageEndpoint endpoint : this.endpoints.values()) {
-			this.activateEndpoint(endpoint);
+		Set<String> endpointNames = this.endpointRegistry.getEndpointNames();
+		for (String name : endpointNames) {
+			MessageEndpoint endpoint = this.endpointRegistry.lookupEndpoint(name);
+			if (endpoint != null) {
+				this.activateEndpoint(endpoint);
+			}
 		}
 	}
 

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<String, MessageChannel> channels = new ConcurrentHashMap<String, MessageChannel>();
+	private final Map<String, MessageChannel> channels = new ConcurrentHashMap<String, MessageChannel>();
 
-	private MessageChannel errorChannel;
+	private volatile MessageChannel errorChannel;
 
 
 	public void setErrorChannel(MessageChannel errorChannel) {

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) {

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<List<MessageHandler>> handlerLists = this.scheduledHandlers.values();
+		for (List<MessageHandler> handlers : handlerLists) {
+			removed = (removed || handlers.remove(handler));
+		}
+		return removed;
+	}
+
 	public boolean isRunning() {
 		return this.running;
 	}

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

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 + "]";
+	}
+
 }

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<String, MessageEndpoint> endpoints = new ConcurrentHashMap<String, MessageEndpoint>();
+
+
+	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<String> getEndpointNames() {
+		return this.endpoints.keySet();
+	}
+
+}

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<String> getEndpointNames();
+
+}

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<String, MessageEndpoint> endpoints = (Map<String, MessageEndpoint>) 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

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 {
+	}
+
 }