From 77e1216559279b835301d635f021f97fc10bd412 Mon Sep 17 00:00:00 2001
From: Gary Russell <grussell@pivotal.io>
Date: Fri, 15 Feb 2019 17:06:06 -0500
Subject: [PATCH] GH-905: Configurable executor on @RabbitListener

Resolves https://github.com/spring-projects/spring-amqp/issues/905
---
 .../test/RabbitListenerTestHarness.java       |  7 ++--
 .../rabbit/annotation/RabbitListener.java     | 10 ++++-
 ...itListenerAnnotationBeanPostProcessor.java | 42 +++++++++++++++----
 ...bstractRabbitListenerContainerFactory.java | 10 ++---
 .../AbstractRabbitListenerEndpoint.java       | 19 ++++++++-
 .../listener/RabbitListenerEndpoint.java      | 14 ++++++-
 .../EnableRabbitIntegrationTests.java         | 13 ++++--
 src/reference/asciidoc/amqp.adoc              |  3 ++
 src/reference/asciidoc/whats-new.adoc         |  4 ++
 9 files changed, 99 insertions(+), 23 deletions(-)

diff --git a/spring-rabbit-test/src/main/java/org/springframework/amqp/rabbit/test/RabbitListenerTestHarness.java b/spring-rabbit-test/src/main/java/org/springframework/amqp/rabbit/test/RabbitListenerTestHarness.java
index 552c06dd..a58f03ed 100644
--- a/spring-rabbit-test/src/main/java/org/springframework/amqp/rabbit/test/RabbitListenerTestHarness.java
+++ b/spring-rabbit-test/src/main/java/org/springframework/amqp/rabbit/test/RabbitListenerTestHarness.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2016 the original author or authors.
+ * Copyright 2016-2019 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.
@@ -69,7 +69,8 @@ public class RabbitListenerTestHarness extends RabbitListenerAnnotationBeanPostP
 
 	@Override
 	protected void processListener(MethodRabbitListenerEndpoint endpoint, RabbitListener rabbitListener, Object bean,
-			Object adminTarget, String beanName) {
+			Object target, String beanName) {
+
 		Object proxy = bean;
 		String id = rabbitListener.id();
 		if (StringUtils.hasText(id)) {
@@ -95,7 +96,7 @@ public class RabbitListenerTestHarness extends RabbitListenerAnnotationBeanPostP
 		else {
 			logger.info("The test harness can only proxy @RabbitListeners with an 'id' attribute");
 		}
-		super.processListener(endpoint, rabbitListener, proxy, adminTarget, beanName);
+		super.processListener(endpoint, rabbitListener, proxy, target, beanName);
 	}
 
 	public InvocationData getNextInvocationDataFor(String id, long wait, TimeUnit unit) throws InterruptedException {
diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/RabbitListener.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/RabbitListener.java
index b981ca23..2bee5432 100644
--- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/RabbitListener.java
+++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/RabbitListener.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2018 the original author or authors.
+ * Copyright 2014-2019 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.
@@ -245,4 +245,12 @@ public @interface RabbitListener {
 	 */
 	String autoStartup() default "";
 
+	/**
+	 * Set the task executor bean name to use for this listener's container; overrides
+	 * any executor set on the container factory.
+	 * @return the executor bean name.
+	 * @since 2.2
+	 */
+	String executor() default "";
+
 }
diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/RabbitListenerAnnotationBeanPostProcessor.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/RabbitListenerAnnotationBeanPostProcessor.java
index c7bd8c2f..0888df5a 100644
--- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/RabbitListenerAnnotationBeanPostProcessor.java
+++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/RabbitListenerAnnotationBeanPostProcessor.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2018 the original author or authors.
+ * Copyright 2014-2019 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.
@@ -69,6 +69,7 @@ import org.springframework.core.annotation.AnnotationUtils;
 import org.springframework.core.convert.ConversionService;
 import org.springframework.core.convert.support.DefaultConversionService;
 import org.springframework.core.env.Environment;
+import org.springframework.core.task.TaskExecutor;
 import org.springframework.lang.Nullable;
 import org.springframework.messaging.handler.annotation.support.DefaultMessageHandlerMethodFactory;
 import org.springframework.messaging.handler.annotation.support.MessageHandlerMethodFactory;
@@ -342,6 +343,7 @@ public class RabbitListenerAnnotationBeanPostProcessor
 
 	private void processMultiMethodListeners(RabbitListener[] classLevelListeners, Method[] multiMethods,
 			Object bean, String beanName) {
+
 		List<Method> checkedMethods = new ArrayList<Method>();
 		Method defaultMethod = null;
 		for (Method method : multiMethods) {
@@ -401,7 +403,8 @@ public class RabbitListenerAnnotationBeanPostProcessor
 	}
 
 	protected void processListener(MethodRabbitListenerEndpoint endpoint, RabbitListener rabbitListener, Object bean,
-			Object adminTarget, String beanName) {
+			Object target, String beanName) {
+
 		endpoint.setBean(bean);
 		endpoint.setMessageHandlerMethodFactory(this.messageHandlerMethodFactory);
 		endpoint.setId(getEndpointId(rabbitListener));
@@ -447,8 +450,9 @@ public class RabbitListenerAnnotationBeanPostProcessor
 			}
 		}
 
-		resolveAdmin(endpoint, rabbitListener, adminTarget);
-		RabbitListenerContainerFactory<?> factory = resolveContainerFactory(rabbitListener, adminTarget, beanName);
+		endpoint.setTaskExecutor(resolveExecutor(rabbitListener, target, beanName));
+		resolveAdmin(endpoint, rabbitListener, target);
+		RabbitListenerContainerFactory<?> factory = resolveContainerFactory(rabbitListener, target, beanName);
 
 		this.registrar.registerEndpoint(endpoint, factory);
 	}
@@ -469,8 +473,9 @@ public class RabbitListenerAnnotationBeanPostProcessor
 	}
 
 	@Nullable
-	private RabbitListenerContainerFactory<?> resolveContainerFactory(RabbitListener rabbitListener, Object adminTarget,
-			String beanName) {
+	private RabbitListenerContainerFactory<?> resolveContainerFactory(RabbitListener rabbitListener,
+			Object factoryTarget, String beanName) {
+
 		RabbitListenerContainerFactory<?> factory = null;
 		String containerFactoryBeanName = resolve(rabbitListener.containerFactory());
 		if (StringUtils.hasText(containerFactoryBeanName)) {
@@ -479,14 +484,33 @@ public class RabbitListenerAnnotationBeanPostProcessor
 				factory = this.beanFactory.getBean(containerFactoryBeanName, RabbitListenerContainerFactory.class);
 			}
 			catch (NoSuchBeanDefinitionException ex) {
-				throw new BeanInitializationException("Could not register rabbit listener endpoint on [" +
-						adminTarget + "] for bean " + beanName + ", no " + RabbitListenerContainerFactory.class.getSimpleName() + " with id '" +
-						containerFactoryBeanName + "' was found in the application context", ex);
+				throw new BeanInitializationException("Could not register rabbit listener endpoint on ["
+						+ factoryTarget + "] for bean " + beanName + ", no "
+						+ RabbitListenerContainerFactory.class.getSimpleName() + " with id '"
+						+ containerFactoryBeanName + "' was found in the application context", ex);
 			}
 		}
 		return factory;
 	}
 
+	@Nullable
+	private TaskExecutor resolveExecutor(RabbitListener rabbitListener, Object execTarget, String beanName) {
+		TaskExecutor exec = null;
+		String execBeanName = resolve(rabbitListener.executor());
+		if (StringUtils.hasText(execBeanName)) {
+			Assert.state(this.beanFactory != null, "BeanFactory must be set to obtain container factory by bean name");
+			try {
+				exec = this.beanFactory.getBean(execBeanName, TaskExecutor.class);
+			}
+			catch (NoSuchBeanDefinitionException ex) {
+				throw new BeanInitializationException("Could not register rabbit listener endpoint on ["
+						+ execTarget + "] for bean " + beanName + ", no " + TaskExecutor.class.getSimpleName()
+						+ " with id '" + execBeanName + "' was found in the application context", ex);
+			}
+		}
+		return exec;
+	}
+
 	private String getEndpointId(RabbitListener rabbitListener) {
 		if (StringUtils.hasText(rabbitListener.id())) {
 			return resolve(rabbitListener.id());
diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/AbstractRabbitListenerContainerFactory.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/AbstractRabbitListenerContainerFactory.java
index 5f43de2c..b12d7e43 100644
--- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/AbstractRabbitListenerContainerFactory.java
+++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/AbstractRabbitListenerContainerFactory.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2018 the original author or authors.
+ * Copyright 2014-2019 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.
@@ -381,10 +381,10 @@ public abstract class AbstractRabbitListenerContainerFactory<C extends AbstractM
 			.acceptIfNotNull(this.autoStartup, instance::setAutoStartup)
 			.acceptIfNotNull(this.phase, instance::setPhase)
 			.acceptIfNotNull(this.afterReceivePostProcessors, instance::setAfterReceivePostProcessors);
-		if (endpoint != null) {
-			if (endpoint.getAutoStartup() != null) {
-				instance.setAutoStartup(endpoint.getAutoStartup());
-			}
+		if (endpoint != null) { // endpoint settings overriding default factory settings
+			javaUtils
+				.acceptIfNotNull(endpoint.getAutoStartup(), instance::setAutoStartup)
+				.acceptIfNotNull(endpoint.getTaskExecutor(), instance::setTaskExecutor);
 			instance.setListenerId(endpoint.getId());
 
 			endpoint.setupListenerContainer(instance);
diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractRabbitListenerEndpoint.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractRabbitListenerEndpoint.java
index d774eb29..61e1634b 100644
--- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractRabbitListenerEndpoint.java
+++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractRabbitListenerEndpoint.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2018 the original author or authors.
+ * Copyright 2014-2019 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.
@@ -33,6 +33,7 @@ import org.springframework.beans.factory.config.BeanExpressionContext;
 import org.springframework.beans.factory.config.BeanExpressionResolver;
 import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
 import org.springframework.context.expression.BeanFactoryResolver;
+import org.springframework.core.task.TaskExecutor;
 import org.springframework.expression.BeanResolver;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
@@ -79,6 +80,8 @@ public abstract class AbstractRabbitListenerEndpoint implements RabbitListenerEn
 
 	private MessageConverter messageConverter;
 
+	private TaskExecutor taskExecutor;
+
 	@Override
 	public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
 		this.beanFactory = beanFactory;
@@ -262,6 +265,20 @@ public abstract class AbstractRabbitListenerEndpoint implements RabbitListenerEn
 		this.messageConverter = messageConverter;
 	}
 
+	@Override
+	public TaskExecutor getTaskExecutor() {
+		return this.taskExecutor;
+	}
+
+	/**
+	 * Override the default task executor.
+	 * @param taskExecutor the executor.
+	 * @since 2.2
+	 */
+	public void setTaskExecutor(TaskExecutor taskExecutor) {
+		this.taskExecutor = taskExecutor;
+	}
+
 	@Override
 	public void setupListenerContainer(MessageListenerContainer listenerContainer) {
 		AbstractMessageListenerContainer container = (AbstractMessageListenerContainer) listenerContainer;
diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/RabbitListenerEndpoint.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/RabbitListenerEndpoint.java
index 7200fe6a..803ce1cf 100644
--- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/RabbitListenerEndpoint.java
+++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/RabbitListenerEndpoint.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2018 the original author or authors.
+ * Copyright 2002-2019 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.
@@ -17,6 +17,7 @@
 package org.springframework.amqp.rabbit.listener;
 
 import org.springframework.amqp.support.converter.MessageConverter;
+import org.springframework.core.task.TaskExecutor;
 import org.springframework.lang.Nullable;
 
 /**
@@ -93,4 +94,15 @@ public interface RabbitListenerEndpoint {
 		return null;
 	}
 
+	/**
+	 * Get the task executor to use for this endpoint's listener container.
+	 * Overrides any executor set on the container factory.
+	 * @return the executor.
+	 * @since 2.2
+	 */
+	@Nullable
+	default TaskExecutor getTaskExecutor() {
+		return null;
+	}
+
 }
diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java
index bd34c859..4c9cae27 100644
--- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java
+++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java
@@ -120,6 +120,7 @@ import org.springframework.core.PriorityOrdered;
 import org.springframework.core.convert.ConversionService;
 import org.springframework.core.convert.converter.Converter;
 import org.springframework.core.convert.support.DefaultConversionService;
+import org.springframework.core.task.TaskExecutor;
 import org.springframework.lang.NonNull;
 import org.springframework.messaging.converter.GenericMessageConverter;
 import org.springframework.messaging.handler.annotation.Header;
@@ -246,7 +247,7 @@ public class EnableRabbitIntegrationTests {
 
 	@Test
 	public void autoSimpleDeclare() {
-		assertEquals("FOOX", rabbitTemplate.convertSendAndReceive("test.simple.declare", "foo"));
+		assertEquals("FOOexec1-1", rabbitTemplate.convertSendAndReceive("test.simple.declare", "foo"));
 	}
 
 	@Test
@@ -921,9 +922,10 @@ public class EnableRabbitIntegrationTests {
 			return foo.toUpperCase();
 		}
 
-		@RabbitListener(queuesToDeclare = @Queue(name = "${jjjj:test.simple.declare}", durable = "true"))
+		@RabbitListener(queuesToDeclare = @Queue(name = "${jjjj:test.simple.declare}", durable = "true"),
+				executor = "exec1")
 		public String handleWithSimpleDeclare(String foo) {
-			return foo.toUpperCase() + "X";
+			return foo.toUpperCase() + Thread.currentThread().getName();
 		}
 
 		@RabbitListener(queuesToDeclare = @Queue, id = "anonymousQueue575")
@@ -1539,6 +1541,11 @@ public class EnableRabbitIntegrationTests {
 			return new TxServiceImpl();
 		}
 
+		@Bean
+		public TaskExecutor exec1() {
+			return new ThreadPoolTaskExecutor();
+		}
+
 		// Rabbit infrastructure setup
 
 		@Bean
diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc
index fc83d021..495a6b3f 100644
--- a/src/reference/asciidoc/amqp.adoc
+++ b/src/reference/asciidoc/amqp.adoc
@@ -2194,6 +2194,9 @@ Its meaning and allowed values depend on the container type, as follows:
 In either case, this setting overrides the settings on the factory.
 Previously you had to define different container factories if you had listeners that required different concurrency.
 
+The annotation also allows overriding the factory `autoStartup` and `taskExecutor` properties via the `autoStartup` and `executor` (since 2.2) annotation properties.
+Using a different executor for each might help with identifying threads associated with each listener in logs and thread dumps.
+
 [[async-annotation-conversion]]
 ====== Message Conversion for Annotated Methods
 
diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc
index 975e7e42..69eac360 100644
--- a/src/reference/asciidoc/whats-new.adoc
+++ b/src/reference/asciidoc/whats-new.adoc
@@ -5,3 +5,7 @@
 
 This section describes the changes between version 2.1 and version 2.2.
 
+===== @RabbitListener Changes
+
+You can now configure an `executor` on each listener, overriding the factory configuration, to more easily identify threads associated with the listener.
+See <<async-annotation-driven-enable>> for more information.