AMQP-811: Listener container factory improvements

JIRA: https://jira.spring.io/browse/AMQP-811

- `ChannelAwareMessageListener` now inherits from `MessageListener`
  - enables `SimpleRabbitListenerEndpoint` to be used with either
  - deprecate setters for CAML
- Support creating a container with no listener

Polishing - PR comments; remove onMessage(message) from AAML -
it was never called since CAML is always preferred by containers.

* Polishing for some stale JavaDocs
* Polishing for the `AsyncRabbitTemplate` to use diamonds and
don't use `rawtypes`
* Fix typo in the `amqp.adoc`

Author: garyrussell

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/AsyncRabbitTemplate.java

@@ -101,11 +101,9 @@ public class AsyncRabbitTemplate implements AsyncAmqpTemplate, ChannelAwareMessa
 
 	private final String replyAddress;
 
-	@SuppressWarnings("rawtypes")
-	private final ConcurrentMap<String, RabbitFuture> pending = new ConcurrentHashMap<String, RabbitFuture>();
+	private final ConcurrentMap<String, RabbitFuture<?>> pending = new ConcurrentHashMap<>();
 
-	@SuppressWarnings("rawtypes")
-	private final CorrelationMessagePostProcessor messagePostProcessor = new CorrelationMessagePostProcessor<>();
+	private final CorrelationMessagePostProcessor<?> messagePostProcessor = new CorrelationMessagePostProcessor<>();
 
 	private volatile boolean running;
 
@@ -224,7 +222,7 @@ public AsyncRabbitTemplate(ConnectionFactory connectionFactory, String exchange,
 		this.container = null;
 		this.replyAddress = null;
 		this.directReplyToContainer = new DirectReplyToMessageListenerContainer(this.template.getConnectionFactory());
-		this.directReplyToContainer.setChannelAwareMessageListener(this);
+		this.directReplyToContainer.setMessageListener(this);
 	}
 
 	/**
@@ -239,7 +237,7 @@ public AsyncRabbitTemplate(RabbitTemplate template) {
 		this.container = null;
 		this.replyAddress = null;
 		this.directReplyToContainer = new DirectReplyToMessageListenerContainer(this.template.getConnectionFactory());
-		this.directReplyToContainer.setChannelAwareMessageListener(this);
+		this.directReplyToContainer.setMessageListener(this);
 	}
 
 	/**

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/AbstractRabbitListenerContainerFactory.java

@@ -353,10 +353,7 @@ public C createListenerContainer(RabbitListenerEndpoint endpoint) {
 		if (this.applicationEventPublisher != null) {
 			instance.setApplicationEventPublisher(this.applicationEventPublisher);
 		}
-		if (endpoint.getAutoStartup() != null) {
-			instance.setAutoStartup(endpoint.getAutoStartup());
-		}
-		else if (this.autoStartup != null) {
+		if (this.autoStartup != null) {
 			instance.setAutoStartup(this.autoStartup);
 		}
 		if (this.phase != null) {
@@ -365,9 +362,14 @@ else if (this.autoStartup != null) {
 		if (this.afterReceivePostProcessors != null) {
 			instance.setAfterReceivePostProcessors(this.afterReceivePostProcessors);
 		}
-		instance.setListenerId(endpoint.getId());
+		if (endpoint != null) {
+			if (endpoint.getAutoStartup() != null) {
+				instance.setAutoStartup(endpoint.getAutoStartup());
+			}
+			instance.setListenerId(endpoint.getId());
 
-		endpoint.setupListenerContainer(instance);
+			endpoint.setupListenerContainer(instance);
+		}
 		if (this.beforeSendReplyPostProcessors != null
 				&& instance.getMessageListener() instanceof AbstractAdaptableMessageListener) {
 			((AbstractAdaptableMessageListener) instance.getMessageListener())

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/DirectRabbitListenerContainerFactory.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2017 the original author or authors.
+ * Copyright 2016-2018 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.
@@ -83,7 +83,7 @@ protected void initializeContainer(DirectMessageListenerContainer instance, Rabb
 		if (this.monitorInterval != null) {
 			instance.setMonitorInterval(this.monitorInterval);
 		}
-		if (endpoint.getConcurrency() != null) {
+		if (endpoint != null && endpoint.getConcurrency() != null) {
 			try {
 				instance.setConsumersPerQueue(Integer.parseInt(endpoint.getConcurrency()));
 			}

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ListenerContainerFactoryBean.java

@@ -22,6 +22,7 @@
 import org.aopalliance.aop.Advice;
 
 import org.springframework.amqp.core.AcknowledgeMode;
+import org.springframework.amqp.core.MessageListener;
 import org.springframework.amqp.core.MessagePostProcessor;
 import org.springframework.amqp.core.Queue;
 import org.springframework.amqp.rabbit.connection.ConnectionFactory;
@@ -82,7 +83,7 @@ public class ListenerContainerFactoryBean extends AbstractFactoryBean<AbstractMe
 
 	private Boolean exposeListenerChannel;
 
-	private Object messageListener;
+	private MessageListener messageListener;
 
 	private ErrorHandler errorHandler;
 
@@ -211,7 +212,7 @@ public void setExposeListenerChannel(boolean exposeListenerChannel) {
 		this.exposeListenerChannel = exposeListenerChannel;
 	}
 
-	public void setMessageListener(Object messageListener) {
+	public void setMessageListener(MessageListener messageListener) {
 		this.messageListener = messageListener;
 	}
 

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/SimpleRabbitListenerContainerFactory.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2017 the original author or authors.
+ * Copyright 2014-2018 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.
@@ -141,11 +141,14 @@ protected void initializeContainer(SimpleMessageListenerContainer instance, Rabb
 		if (this.txSize != null) {
 			instance.setTxSize(this.txSize);
 		}
-		String concurrency = endpoint.getConcurrency();
-		if (concurrency != null) {
-			instance.setConcurrency(concurrency);
+		String concurrency = null;
+		if (endpoint != null) {
+			concurrency = endpoint.getConcurrency();
+			if (concurrency != null) {
+				instance.setConcurrency(concurrency);
+			}
 		}
-		else if (this.concurrentConsumers != null) {
+		if (concurrency == null && this.concurrentConsumers != null) {
 			instance.setConcurrentConsumers(this.concurrentConsumers);
 		}
 		if ((concurrency == null || !(concurrency.contains("-"))) && this.maxConcurrentConsumers != null) {

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java

@@ -394,23 +394,23 @@ public void setExposeListenerChannel(boolean exposeListenerChannel) {
 	/**
 	 * Set the message listener implementation to register. This can be either a Spring
 	 * {@link MessageListener} object or a Spring {@link ChannelAwareMessageListener}
-	 * object. Using the strongly typed
-	 * {@link #setChannelAwareMessageListener(ChannelAwareMessageListener)} is preferred.
+	 * object.
 	 *
 	 * @param messageListener The listener.
 	 * @throws IllegalArgumentException if the supplied listener is not a
 	 * {@link MessageListener} or a {@link ChannelAwareMessageListener}
+	 * @deprecated use {@link #setMessageListener(MessageListener)}.
 	 * @see MessageListener
 	 * @see ChannelAwareMessageListener
 	 */
+	@Deprecated
 	public void setMessageListener(Object messageListener) {
 		checkMessageListener(messageListener);
 		this.messageListener = messageListener;
 	}
 
 	/**
-	 * Set the {@link MessageListener}; strongly typed version of
-	 * {@link #setMessageListener(Object)}.
+	 * Set the {@link MessageListener}.
 	 * @param messageListener the listener.
 	 * @since 2.0
 	 */
@@ -419,27 +419,27 @@ public void setMessageListener(MessageListener messageListener) {
 	}
 
 	/**
-	 * Set the {@link ChannelAwareMessageListener}; strongly typed version of
-	 * {@link #setMessageListener(Object)}.
+	 * Set the {@link ChannelAwareMessageListener}.
 	 * @param messageListener the listener.
 	 * @since 2.0
+	 * @deprecated use {@link #setMessageListener(MessageListener)} since
+	 * {@link ChannelAwareMessageListener} now inherits {@link MessageListener}.
 	 */
+	@Deprecated
 	public void setChannelAwareMessageListener(ChannelAwareMessageListener messageListener) {
 		this.messageListener = messageListener;
 	}
 
 	/**
 	 * Check the given message listener, throwing an exception if it does not correspond to a supported listener type.
 	 * <p>
-	 * By default, only a Spring {@link MessageListener} object or a Spring
-	 * {@link ChannelAwareMessageListener} object will be accepted.
+	 * Only a Spring {@link MessageListener} object will be accepted.
 	 * @param messageListener the message listener object to check
-	 * @throws IllegalArgumentException if the supplied listener is not a MessageListener or SessionAwareMessageListener
+	 * @throws IllegalArgumentException if the supplied listener is not a MessageListener
 	 * @see MessageListener
-	 * @see ChannelAwareMessageListener
 	 */
 	protected void checkMessageListener(Object messageListener) {
-		if (!(messageListener instanceof MessageListener || messageListener instanceof ChannelAwareMessageListener)) {
+		if (!(messageListener instanceof MessageListener)) {
 			throw new IllegalArgumentException("Message listener needs to be of type ["
 					+ MessageListener.class.getName() + "] or [" + ChannelAwareMessageListener.class.getName() + "]");
 		}
@@ -1071,7 +1071,7 @@ public final void afterPropertiesSet() {
 	}
 
 	@Override
-	public void setupMessageListener(Object messageListener) {
+	public void setupMessageListener(MessageListener messageListener) {
 		setMessageListener(messageListener);
 	}
 
@@ -1392,7 +1392,7 @@ protected void invokeListener(Channel channel, Message message) throws Exception
 	 * @param channel the Rabbit Channel to operate on
 	 * @param message the received Rabbit Message
 	 * @throws Exception if thrown by Rabbit API methods
-	 * @see #setMessageListener
+	 * @see #setMessageListener(MessageListener)
 	 */
 	protected void actualInvokeListener(Channel channel, Message message) throws Exception {
 		Object listener = getMessageListener();

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectReplyToMessageListenerContainer.java

@@ -82,33 +82,42 @@ public final boolean removeQueueNames(String... queueNames) {
 	}
 
 	@Override
+	@SuppressWarnings("deprecation")
+	@Deprecated
 	public void setMessageListener(Object messageListener) {
 		throw new UnsupportedOperationException(
 				"'messageListener' must be a 'MessageListener' or 'ChannelAwareMessageListener'");
 	}
 
 	@Override
+	@SuppressWarnings("deprecation")
+	@Deprecated
 	public void setChannelAwareMessageListener(ChannelAwareMessageListener messageListener) {
-		super.setChannelAwareMessageListener((message, channel) -> {
-			try {
-				messageListener.onMessage(message, channel);
-			}
-			finally {
-				this.inUseConsumerChannels.remove(channel);
-			}
-		});
+		setMessageListener(messageListener);
 	}
 
 	@Override
 	public void setMessageListener(MessageListener messageListener) {
-		super.setChannelAwareMessageListener((message, channel) -> {
-			try {
-				messageListener.onMessage(message);
-			}
-			finally {
-				this.inUseConsumerChannels.remove(channel);
-			}
-		});
+		if (messageListener instanceof ChannelAwareMessageListener) {
+			super.setMessageListener((ChannelAwareMessageListener) (message, channel) -> {
+				try {
+					((ChannelAwareMessageListener) messageListener).onMessage(message, channel);
+				}
+				finally {
+					this.inUseConsumerChannels.remove(channel);
+				}
+			});
+		}
+		else {
+			super.setMessageListener((ChannelAwareMessageListener) (message, channel) -> {
+				try {
+					messageListener.onMessage(message);
+				}
+				finally {
+					this.inUseConsumerChannels.remove(channel);
+				}
+			});
+		}
 	}
 
 	@Override

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/MessageListenerContainer.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014 the original author or authors.
+ * Copyright 2014-2018 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.
@@ -16,6 +16,7 @@
 
 package org.springframework.amqp.rabbit.listener;
 
+import org.springframework.amqp.core.MessageListener;
 import org.springframework.amqp.support.converter.MessageConverter;
 import org.springframework.context.SmartLifecycle;
 
@@ -33,7 +34,7 @@ public interface MessageListenerContainer extends SmartLifecycle {
 	 * if that message listener type is not supported.
 	 * @param messageListener the {@code object} to wrapped to the {@code MessageListener}.
 	 */
-	void setupMessageListener(Object messageListener);
+	void setupMessageListener(MessageListener messageListener);
 
 	/**
 	 * @return the {@link MessageConverter} that can be used to

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/RabbitListenerContainerFactory.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2017 the original author or authors.
+ * Copyright 2002-2018 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.
@@ -16,11 +16,11 @@
 
 package org.springframework.amqp.rabbit.listener;
 
-
+import org.springframework.amqp.core.MessageListener;
+import org.springframework.lang.Nullable;
 
 /**
- * Factory of {@link MessageListenerContainer} based on a
- * {@link RabbitListenerEndpoint} definition.
+ * Factory of {@link MessageListenerContainer}s.
  * @param <C> the container type.
  * @author Stephane Nicoll
  * @author Gary Russell
@@ -31,10 +31,21 @@
 public interface RabbitListenerContainerFactory<C extends MessageListenerContainer> {
 
 	/**
-	 * Create a {@link MessageListenerContainer} for the given {@link RabbitListenerEndpoint}.
-	 * @param endpoint the endpoint to configure
-	 * @return the created container
+	 * Create a {@link MessageListenerContainer} for the given
+	 * {@link RabbitListenerEndpoint}.
+	 * @param endpoint the endpoint to configure.
+	 * @return the created container.
+	 */
+	C createListenerContainer(@Nullable RabbitListenerEndpoint endpoint);
+
+	/**
+	 * Create a {@link MessageListenerContainer} with no {@link MessageListener}
+	 * or queues; the listener must be added later before the container is started.
+	 * @return the created container.
+	 * @since 2.1.
 	 */
-	C createListenerContainer(RabbitListenerEndpoint endpoint);
+	default C createListenerContainer() {
+		return createListenerContainer(null);
+	}
 
 }

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/AbstractAdaptableMessageListener.java

@@ -29,7 +29,6 @@
 import org.springframework.amqp.core.MessagePostProcessor;
 import org.springframework.amqp.core.MessageProperties;
 import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
-import org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException;
 import org.springframework.amqp.rabbit.support.DefaultMessagePropertiesConverter;
 import org.springframework.amqp.rabbit.support.MessagePropertiesConverter;
 import org.springframework.amqp.rabbit.support.RabbitExceptionTranslator;
@@ -58,10 +57,9 @@
  *
  * @since 1.4
  *
- * @see MessageListener
  * @see ChannelAwareMessageListener
  */
-public abstract class AbstractAdaptableMessageListener implements MessageListener, ChannelAwareMessageListener {
+public abstract class AbstractAdaptableMessageListener implements ChannelAwareMessageListener {
 
 	private static final String DEFAULT_RESPONSE_ROUTING_KEY = "";
 
@@ -219,27 +217,6 @@ protected MessageConverter getMessageConverter() {
 		return this.messageConverter;
 	}
 
-	/**
-	 * Rabbit {@link MessageListener} entry point.
-	 * <p>
-	 * Delegates the message to the target listener method, with appropriate conversion of the message argument.
-	 * <p>
-	 * <b>Note:</b> Does not support sending response messages based on result objects returned from listener methods.
-	 * Use the {@link ChannelAwareMessageListener} entry point (typically through a Spring message listener container)
-	 * for handling result objects as well.
-	 * @param message the incoming Rabbit message
-	 * @see #onMessage(Message, com.rabbitmq.client.Channel)
-	 */
-	@Override
-	public void onMessage(Message message) {
-		try {
-			onMessage(message, null);
-		}
-		catch (Exception e) {
-			throw new ListenerExecutionFailedException("Listener threw exception", e, message);
-		}
-	}
-
 	/**
 	 * Handle the given exception that arose during listener execution.
 	 * The default implementation logs the exception at error level.