GH-1140: RabbitTemplate and CGLIB Proxies

Resolves #1140

`start()` and `stop()` methods are `final`, causing NPEs with CGLIB
proxies.

- remove `final` modifiers
- pull `Lifecycle` up to `RabbitOperations` so JDK proxies can be used
- add default no-op implementations for backward compatibility

Also implement `DisposableBean` so that reply containers are stopped.

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitOperations.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2020 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.
@@ -22,6 +22,7 @@
 import org.springframework.amqp.core.MessagePostProcessor;
 import org.springframework.amqp.rabbit.connection.ConnectionFactory;
 import org.springframework.amqp.rabbit.connection.CorrelationData;
+import org.springframework.context.Lifecycle;
 import org.springframework.core.ParameterizedTypeReference;
 import org.springframework.lang.Nullable;
 
@@ -33,7 +34,7 @@
  * @author Gary Russell
  * @author Artem Bilan
  */
-public interface RabbitOperations extends AmqpTemplate {
+public interface RabbitOperations extends AmqpTemplate, Lifecycle {
 
 	/**
 	 * Execute the callback with a channel and reliably close the channel afterwards.
@@ -421,6 +422,22 @@ <T> T convertSendAndReceiveAsType(String exchange, String routingKey, Object mes
 			@Nullable CorrelationData correlationData,
 			ParameterizedTypeReference<T> responseType) throws AmqpException;
 
+
+	@Override
+	default void start() {
+		// No-op - implemented for backward compatibility
+	}
+
+	@Override
+	default void stop() {
+		// No-op - implemented for backward compatibility
+	}
+
+	@Override
+	default boolean isRunning() {
+		return false;
+	}
+
 	/**
 	 * Callback for using the same channel for multiple RabbitTemplate
 	 * operations.

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2020 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.
@@ -80,7 +80,7 @@
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.BeanFactoryAware;
 import org.springframework.beans.factory.BeanNameAware;
-import org.springframework.context.Lifecycle;
+import org.springframework.beans.factory.DisposableBean;
 import org.springframework.context.expression.BeanFactoryResolver;
 import org.springframework.context.expression.MapAccessor;
 import org.springframework.core.ParameterizedTypeReference;
@@ -149,7 +149,7 @@
  */
 public class RabbitTemplate extends RabbitAccessor // NOSONAR type line count
 		implements BeanFactoryAware, RabbitOperations, MessageListener,
-		ListenerContainerAware, PublisherCallbackChannel.Listener, Lifecycle, BeanNameAware {
+		ListenerContainerAware, PublisherCallbackChannel.Listener, BeanNameAware, DisposableBean {
 
 	private static final String UNCHECKED = "unchecked";
 
@@ -879,7 +879,7 @@ public int getUnconfirmedCount() {
 	}
 
 	@Override
-	public final void start() {
+	public void start() {
 		doStart();
 	}
 
@@ -892,7 +892,7 @@ protected void doStart() {
 	}
 
 	@Override
-	public final void stop() {
+	public void stop() {
 		synchronized (this.directReplyToContainers) {
 			this.directReplyToContainers.values()
 					.stream()
@@ -920,6 +920,11 @@ public boolean isRunning() {
 		}
 	}
 
+	@Override
+	public void destroy() {
+		stop();
+	}
+
 	private void evaluateFastReplyTo() {
 		this.usingFastReplyTo = useDirectReplyTo();
 		this.evaluatedFastReplyTo = true;