Fix missing class-level @SendTo for CGLib proxy

Similar to spring-projects/spring-kafka#1631

The `DelegatingInvocableHandler` doesn't take into account a possible proxy
nature of the listener class.

* Fix `DelegatingInvocableHandler.setupReplyTo()` to deal with user class
which is origin for possible CGLib proxy for the proper annotations search

**Cherry-pick to 2.2.x**

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2019 the original author or authors.
+ * Copyright 2015-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.
@@ -164,8 +164,9 @@ private void setupReplyTo(InvocableHandlerMethod handler) {
 			replyTo = extractSendTo(method.toString(), ann);
 		}
 		if (replyTo == null) {
-			SendTo ann = AnnotationUtils.getAnnotation(this.bean.getClass(), SendTo.class);
-			replyTo = extractSendTo(getBean().getClass().getSimpleName(), ann);
+			Class<?> beanType = handler.getBeanType();
+			SendTo ann = AnnotationUtils.getAnnotation(beanType, SendTo.class);
+			replyTo = extractSendTo(beanType.getSimpleName(), ann);
 		}
 		if (replyTo != null) {
 			this.handlerSendTo.put(handler, PARSER.parseExpression(replyTo, PARSER_CONTEXT));

spring-rabbit/src/test/kotlin/org/springframework/amqp/rabbit/annotation/EnableRabbitKotlinTests.kt

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2019 the original author or authors.
+ * Copyright 2018-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.
@@ -17,6 +17,7 @@
 package org.springframework.amqp.rabbit.annotation
 
 import assertk.assertThat
+import assertk.assertions.isEqualTo
 import assertk.assertions.isTrue
 import org.junit.jupiter.api.Test
 import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory
@@ -25,9 +26,13 @@ import org.springframework.amqp.rabbit.core.RabbitTemplate
 import org.springframework.amqp.rabbit.junit.RabbitAvailable
 import org.springframework.amqp.rabbit.junit.RabbitAvailableCondition
 import org.springframework.amqp.rabbit.listener.api.RabbitListenerErrorHandler
+import org.springframework.aop.framework.ProxyFactory
+import org.springframework.beans.BeansException
 import org.springframework.beans.factory.annotation.Autowired
+import org.springframework.beans.factory.config.BeanPostProcessor
 import org.springframework.context.annotation.Bean
 import org.springframework.context.annotation.Configuration
+import org.springframework.messaging.handler.annotation.SendTo
 import org.springframework.test.annotation.DirtiesContext
 import org.springframework.test.context.junit.jupiter.SpringJUnitConfig
 import java.util.concurrent.CountDownLatch
@@ -37,12 +42,13 @@ import java.util.concurrent.TimeUnit
  * Kotlin Annotated listener tests.
  *
  * @author Gary Russell
+ * @author Artem Bilan
  *
  * @since 2.1
  *
  */
 @SpringJUnitConfig
-@RabbitAvailable(queues = ["kotlinQueue", "kotlinQueue1"])
+@RabbitAvailable(queues = ["kotlinQueue", "kotlinQueue1", "kotlinReplyQueue"])
 @DirtiesContext
 class EnableRabbitKotlinTests {
 
@@ -61,6 +67,8 @@ class EnableRabbitKotlinTests {
 		val template = RabbitTemplate(this.config.cf())
 		template.convertAndSend("kotlinQueue1", "test")
 		assertThat(this.config.ehLatch.await(10, TimeUnit.SECONDS)).isTrue();
+		val reply = template.receiveAndConvert("kotlinReplyQueue")
+		assertThat(reply).isEqualTo("error processed");
 	}
 
 	@Configuration
@@ -92,18 +100,34 @@ class EnableRabbitKotlinTests {
 			return Multi()
 		}
 
+		@Bean
+		fun proxyListenerPostProcessor(): BeanPostProcessor? {
+			return object : BeanPostProcessor {
+				@Throws(BeansException::class)
+				override fun postProcessBeforeInitialization(bean: Any, beanName: String): Any {
+					if (bean is Multi) {
+						val proxyFactory = ProxyFactory(bean)
+						proxyFactory.isProxyTargetClass = true
+						return proxyFactory.proxy
+					}
+					return bean
+				}
+			}
+		}
+
 		val ehLatch = CountDownLatch(1)
 
 		@Bean
 		fun eh() = RabbitListenerErrorHandler { _, _, _ ->
 			this.ehLatch.countDown()
-			null
+			"error processed"
 		}
 
 	}
 
 	@RabbitListener(queues = ["kotlinQueue1"], errorHandler = "#{eh}")
-	class Multi {
+	@SendTo("kotlinReplyQueue")
+	open class Multi {
 
 		@RabbitHandler
 		fun handle(@Suppress("UNUSED_PARAMETER") data: String) {