GH-3014: Add request/reply support into RabbitAmqpMessageListenerAdapter

Fixes: #3014

Author: artembilan

build.gradle

@@ -481,6 +481,7 @@ project('spring-rabbitmq-client') {
 		api "com.rabbitmq.client:amqp-client:$rabbitmqAmqpClientVersion"
 
 		testApi project(':spring-rabbit-junit')
+		testApi 'io.projectreactor:reactor-core'
 
 		testRuntimeOnly 'com.fasterxml.jackson.core:jackson-databind'
 

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

@@ -216,6 +216,10 @@ public void setBeforeSendReplyPostProcessors(MessagePostProcessor... beforeSendR
 				beforeSendReplyPostProcessors.length);
 	}
 
+	public MessagePostProcessor @Nullable [] getBeforeSendReplyPostProcessors() {
+		return this.beforeSendReplyPostProcessors;
+	}
+
 	/**
 	 * Set a {@link RetryTemplate} to use when sending replies.
 	 * @param retryTemplate the template.
@@ -369,7 +373,7 @@ protected void handleResult(InvocationResult resultArg, Message request, @Nullab
 	/**
 	 * Handle the given result object returned from the listener method, sending a
 	 * response message back.
-	 * @param resultArg the result object to handle (never <code>null</code>)
+	 * @param resultArg the result object to handle
 	 * @param request the original request message
 	 * @param channel the Rabbit channel to operate on (maybe <code>null</code>)
 	 * @param source the source data for the method invocation - e.g.
@@ -383,7 +387,7 @@ protected void handleResult(InvocationResult resultArg, Message request, @Nullab
 	protected void handleResult(@Nullable InvocationResult resultArg, Message request,
 			@Nullable Channel channel, @Nullable Object source) {
 
-		if (channel != null && resultArg != null) {
+		if (resultArg != null) {
 			if (resultArg.getReturnValue() instanceof CompletableFuture<?> completable) {
 				if (!this.isManualAck) {
 					this.logger.warn("Container AcknowledgeMode must be MANUAL for a Future<?> return type; "
@@ -413,13 +417,9 @@ else if (monoPresent && MonoHandler.isMono(resultArg.getReturnValue())) {
 				doHandleResult(resultArg, request, channel, source);
 			}
 		}
-		else if (this.logger.isWarnEnabled()) {
-			this.logger.warn("Listener method returned result [" + resultArg
-					+ "]: not generating response message for it because no Rabbit Channel given");
-		}
 	}
 
-	private void asyncSuccess(InvocationResult resultArg, Message request, Channel channel,
+	private void asyncSuccess(InvocationResult resultArg, Message request, @Nullable Channel channel,
 			@Nullable Object source, @Nullable Object deferredResult) {
 
 		if (deferredResult == null) {
@@ -458,8 +458,9 @@ protected void basicAck(Message request, @Nullable Channel channel) {
 		}
 	}
 
-	protected void asyncFailure(Message request, Channel channel, Throwable t, @Nullable Object source) {
+	protected void asyncFailure(Message request, @Nullable Channel channel, Throwable t, @Nullable Object source) {
 		this.logger.error("Future, Mono, or suspend function was completed with an exception for " + request, t);
+		Assert.notNull(channel, "'channel' must not be null.");
 		try {
 			channel.basicNack(request.getMessageProperties().getDeliveryTag(), false,
 					ContainerUtils.shouldRequeue(this.defaultRequeueRejected, t, this.logger));
@@ -469,7 +470,7 @@ protected void asyncFailure(Message request, Channel channel, Throwable t, @Null
 		}
 	}
 
-	protected void doHandleResult(InvocationResult resultArg, Message request, Channel channel,
+	protected void doHandleResult(InvocationResult resultArg, Message request, @Nullable Channel channel,
 			@Nullable Object source) {
 
 		if (this.logger.isDebugEnabled()) {
@@ -500,12 +501,13 @@ protected void doHandleResult(InvocationResult resultArg, Message request, Chann
 	/**
 	 * Build a Rabbit message to be sent as response based on the given result object.
 	 * @param channel the Rabbit Channel to operate on.
+	 *                Can be null if implementation does not support AMQP 0.9.1.
 	 * @param result the content of the message, as returned from the listener method.
 	 * @param genericType the generic type to populate type headers.
 	 * @return the Rabbit <code>Message</code> (never <code>null</code>).
 	 * @see #setMessageConverter
 	 */
-	protected Message buildMessage(Channel channel, @Nullable Object result, @Nullable Type genericType) {
+	protected Message buildMessage(@Nullable Channel channel, @Nullable Object result, @Nullable Type genericType) {
 		MessageConverter converter = getMessageConverter();
 		if (converter != null && !(result instanceof Message)) {
 			return convert(result, genericType, converter);
@@ -633,7 +635,8 @@ private Address evaluateReplyTo(Message request, @Nullable Object source, @Nulla
 	 * @see #postProcessResponse(Message, Message)
 	 * @see #setReplyPostProcessor(ReplyPostProcessor)
 	 */
-	protected void sendResponse(Channel channel, Address replyTo, Message messageIn) {
+	protected void sendResponse(@Nullable Channel channel, Address replyTo, Message messageIn) {
+		Assert.notNull(channel, "'channel' must not be null.");
 		Message message = messageIn;
 		if (this.beforeSendReplyPostProcessors != null) {
 			for (MessagePostProcessor postProcessor : this.beforeSendReplyPostProcessors) {

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

@@ -169,7 +169,7 @@ public void onMessage(org.springframework.amqp.core.Message amqpMessage, @Nullab
 	}
 
 	@Override
-	protected void asyncFailure(org.springframework.amqp.core.Message request, Channel channel, Throwable t,
+	protected void asyncFailure(org.springframework.amqp.core.Message request, @Nullable Channel channel, Throwable t,
 			@Nullable Object source) {
 
 		try {
@@ -183,7 +183,7 @@ protected void asyncFailure(org.springframework.amqp.core.Message request, Chann
 		super.asyncFailure(request, channel, t, source);
 	}
 
-	private void handleException(org.springframework.amqp.core.Message amqpMessage, @Nullable Channel channel,
+	protected void handleException(org.springframework.amqp.core.Message amqpMessage, @Nullable Channel channel,
 			@Nullable Message<?> message, ListenerExecutionFailedException e) throws Exception { // NOSONAR
 
 		if (this.errorHandler != null) {
@@ -307,7 +307,7 @@ private String createMessagingErrorMessage(Object payload) {
 	 * @see #setMessageConverter
 	 */
 	@Override
-	protected org.springframework.amqp.core.Message buildMessage(Channel channel, @Nullable Object result,
+	protected org.springframework.amqp.core.Message buildMessage(@Nullable Channel channel, @Nullable Object result,
 			@Nullable Type genericType) {
 
 		MessageConverter converter = getMessageConverter();

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/adapter/MessageListenerAdapterTests.java

@@ -24,6 +24,7 @@
 import java.util.concurrent.atomic.AtomicReference;
 
 import com.rabbitmq.client.Channel;
+import org.jspecify.annotations.Nullable;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import reactor.core.publisher.Mono;
@@ -33,7 +34,6 @@
 import org.springframework.amqp.core.Message;
 import org.springframework.amqp.core.MessageProperties;
 import org.springframework.amqp.support.SendRetryContextAccessor;
-import org.springframework.amqp.support.converter.SimpleMessageConverter;
 import org.springframework.aop.framework.ProxyFactory;
 import org.springframework.retry.RetryPolicy;
 import org.springframework.retry.policy.SimpleRetryPolicy;
@@ -67,8 +67,15 @@ public class MessageListenerAdapterTests {
 	public void init() {
 		this.messageProperties = new MessageProperties();
 		this.messageProperties.setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN);
-		this.adapter = new MessageListenerAdapter();
-		this.adapter.setMessageConverter(new SimpleMessageConverter());
+		this.adapter = new MessageListenerAdapter() {
+
+			@Override
+			protected void doHandleResult(InvocationResult resultArg, Message request, @Nullable Channel channel,
+					@Nullable Object source) {
+
+			}
+
+		};
 	}
 
 	@Test
@@ -77,7 +84,7 @@ class ExtendedListenerAdapter extends MessageListenerAdapter {
 
 			@Override
 			protected Object[] buildListenerArguments(Object extractedMessage, Channel channel, Message message) {
-				return new Object[] { extractedMessage, channel, message };
+				return new Object[] {extractedMessage, channel, message};
 			}
 
 		}
@@ -131,7 +138,15 @@ public String myPojoMessageMethod(String input) {
 			}
 
 		}
-		this.adapter = new MessageListenerAdapter(new Delegate(), "myPojoMessageMethod");
+		this.adapter = new MessageListenerAdapter(new Delegate(), "myPojoMessageMethod") {
+
+			@Override
+			protected void doHandleResult(InvocationResult resultArg, Message request, @Nullable Channel channel,
+					@Nullable Object source) {
+
+			}
+
+		};
 		this.adapter.onMessage(new Message("foo".getBytes(), messageProperties), null);
 		assertThat(called.get()).isTrue();
 	}
@@ -146,7 +161,7 @@ public void testExplicitListenerMethod() throws Exception {
 
 	@Test
 	public void testMappedListenerMethod() throws Exception {
-		Map<String, String> map = new HashMap<String, String>();
+		Map<String, String> map = new HashMap<>();
 		map.put("foo", "handle");
 		map.put("bar", "notDefinedOnInterface");
 		this.adapter.setDefaultListenerMethod("anotherHandle");
@@ -186,6 +201,7 @@ public void testJdkProxyListener() throws Exception {
 
 	@Test
 	public void testReplyRetry() throws Exception {
+		this.adapter = new MessageListenerAdapter();
 		this.adapter.setDefaultListenerMethod("handle");
 		this.adapter.setDelegate(this.simpleService);
 		RetryPolicy retryPolicy = new SimpleRetryPolicy(2);
@@ -210,7 +226,7 @@ public void testReplyRetry() throws Exception {
 		this.adapter.onMessage(message, channel);
 		assertThat(this.simpleService.called).isEqualTo("handle");
 		assertThat(replyMessage.get()).isNotNull();
-		assertThat(new String(replyMessage.get().getBody())).isEqualTo("processedfoo");
+		assertThat(new String(replyMessage.get().getBody())).isEqualTo("processed foo");
 		assertThat(replyAddress.get()).isNotNull();
 		assertThat(replyAddress.get().getExchangeName()).isEqualTo("foo");
 		assertThat(replyAddress.get().getRoutingKey()).isEqualTo("bar");
@@ -224,7 +240,7 @@ class Delegate {
 			@SuppressWarnings("unused")
 			public CompletableFuture<String> myPojoMessageMethod(String input) {
 				CompletableFuture<String> future = new CompletableFuture<>();
-				future.complete("processed" + input);
+				future.complete("processed " + input);
 				return future;
 			}
 
@@ -270,18 +286,18 @@ public static class SimpleService implements Service {
 		@Override
 		public String handle(String input) {
 			called = "handle";
-			return "processed" + input;
+			return "processed " + input;
 		}
 
 		@Override
 		public String anotherHandle(String input) {
 			called = "anotherHandle";
-			return "processed" + input;
+			return "processed " + input;
 		}
 
 		public String notDefinedOnInterface(String input) {
 			called = "notDefinedOnInterface";
-			return "processed" + input;
+			return "processed " + input;
 		}
 
 	}

spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/listener/RabbitAmqpListenerContainer.java

@@ -99,6 +99,8 @@ public class RabbitAmqpListenerContainer implements MessageListenerContainer, Be
 
 	private @Nullable MessageListener proxy;
 
+	private boolean asyncReplies;
+
 	private ErrorHandler errorHandler = new ConditionalRejectingErrorHandler();
 
 	private @Nullable Collection<MessagePostProcessor> afterReceivePostProcessors;
@@ -255,6 +257,10 @@ public String getListenerId() {
 	@Override
 	public void setupMessageListener(MessageListener messageListener) {
 		this.messageListener = messageListener;
+		this.asyncReplies = messageListener.isAsyncReplies();
+		if (this.messageListener instanceof RabbitAmqpMessageListenerAdapter rabbitAmqpMessageListenerAdapter) {
+			rabbitAmqpMessageListenerAdapter.setConnectionFactory(this.connectionFactory);
+		}
 		this.proxy = this.messageListener;
 		if (!ObjectUtils.isEmpty(this.adviceChain)) {
 			ProxyFactory factory = new ProxyFactory(messageListener);
@@ -276,6 +282,11 @@ public void afterPropertiesSet() {
 		Assert.state(this.queues != null, "At least one queue has to be provided for consuming.");
 		Assert.state(this.messageListener != null, "The 'messageListener' must be provided.");
 
+		if (this.asyncReplies && this.autoSettle) {
+			LOG.info("Enforce MANUAL settlement for async replies.");
+			this.autoSettle = false;
+		}
+
 		this.messageListener.containerAckMode(this.autoSettle ? AcknowledgeMode.AUTO : AcknowledgeMode.MANUAL);
 		if (this.messageListener instanceof RabbitAmqpMessageListenerAdapter adapter
 				&& this.afterReceivePostProcessors != null) {