GH-3002: Add RPC support to RabbitAmqpTemplate

Fixes: https://github.com/spring-projects/spring-amqp/issues/3002

* Implement `sendAndReceive` & `receiveAndReply` operations in the `RabbitAmqpTemplate`
* Expose contracts to the `AsyncAmqpTemplate`
* Some NullAway fixes for `AsyncAmqpTemplate` hierarchy
* Move DLQ objects for testing to the common `RabbitAmqpTestBase`

spring-amqp/src/main/java/org/springframework/amqp/core/AsyncAmqpTemplate.java

@@ -92,11 +92,30 @@ public interface AsyncAmqpTemplate {
 		throw new UnsupportedOperationException();
 	}
 
-	default <T> CompletableFuture<T> receiveAndConvert(ParameterizedTypeReference<T> type) {
+	default <T> CompletableFuture<T> receiveAndConvert(@Nullable ParameterizedTypeReference<T> type) {
 		throw new UnsupportedOperationException();
 	}
 
-	default <T> CompletableFuture<T> receiveAndConvert(String queueName, ParameterizedTypeReference<T> type) {
+	default <T> CompletableFuture<T> receiveAndConvert(String queueName, @Nullable ParameterizedTypeReference<T> type) {
+		throw new UnsupportedOperationException();
+	}
+
+	default <R, S> CompletableFuture<Boolean> receiveAndReply(ReceiveAndReplyCallback<R, S> callback) {
+		throw new UnsupportedOperationException();
+	}
+
+	/**
+	 * Perform a server-side RPC functionality.
+	 * The request message must have a {@code replyTo} property.
+	 * The request {@code messageId} property is used for correlation.
+	 * The callback might not produce a reply with the meaning nothing to answer.
+	 * @param queueName the queue to consume request.
+	 * @param callback an application callback to handle request and produce reply.
+	 * @return the completion status: true if no errors and reply has been produced.
+	 * @param <R> the request body type.
+	 * @param <S> the response body type
+	 */
+	default <R, S> CompletableFuture<Boolean> receiveAndReply(String queueName, ReceiveAndReplyCallback<R, S> callback) {
 		throw new UnsupportedOperationException();
 	}
 
@@ -240,8 +259,9 @@ public interface AsyncAmqpTemplate {
 	 * @param <C> the expected result type.
 	 * @return the {@link CompletableFuture}.
 	 */
-	<C> CompletableFuture<C> convertSendAndReceiveAsType(Object object, MessagePostProcessor messagePostProcessor,
-			ParameterizedTypeReference<C> responseType);
+	<C> CompletableFuture<C> convertSendAndReceiveAsType(Object object,
+			@Nullable MessagePostProcessor messagePostProcessor,
+			@Nullable ParameterizedTypeReference<C> responseType);
 
 	/**
 	 * Convert the object to a message and send it to the default exchange with the

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

@@ -464,7 +464,7 @@ public class AsyncRabbitTemplate implements AsyncAmqpTemplate, ChannelAwareMessa
 
 	@Override
 	public <C> RabbitConverterFuture<C> convertSendAndReceiveAsType(Object object,
-			MessagePostProcessor messagePostProcessor, ParameterizedTypeReference<C> responseType) {
+			@Nullable MessagePostProcessor messagePostProcessor, @Nullable ParameterizedTypeReference<C> responseType) {
 
 		return convertSendAndReceiveAsType(this.template.getExchange(), this.template.getRoutingKey(), object,
 				messagePostProcessor, responseType);

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

@@ -17,25 +17,28 @@
 package org.springframework.amqp.rabbitmq.client;
 
 import java.time.Duration;
+import java.util.Objects;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.Supplier;
 
 import com.rabbitmq.client.amqp.Consumer;
 import com.rabbitmq.client.amqp.Environment;
 import com.rabbitmq.client.amqp.Publisher;
 import com.rabbitmq.client.amqp.Resource;
+import com.rabbitmq.client.amqp.RpcClient;
 import org.jspecify.annotations.Nullable;
 
-import org.springframework.amqp.AmqpException;
+import org.springframework.amqp.AmqpIllegalStateException;
 import org.springframework.amqp.core.AmqpTemplate;
 import org.springframework.amqp.core.AsyncAmqpTemplate;
 import org.springframework.amqp.core.Message;
 import org.springframework.amqp.core.MessagePostProcessor;
 import org.springframework.amqp.core.MessageProperties;
 import org.springframework.amqp.core.ReceiveAndReplyCallback;
-import org.springframework.amqp.core.ReplyToAddressCallback;
+import org.springframework.amqp.core.ReceiveAndReplyMessageCallback;
 import org.springframework.amqp.rabbit.core.AmqpNackReceivedException;
 import org.springframework.amqp.support.converter.MessageConverter;
 import org.springframework.amqp.support.converter.SimpleMessageConverter;
@@ -43,7 +46,9 @@ import org.springframework.amqp.support.converter.SmartMessageConverter;
 import org.springframework.amqp.utils.JavaUtils;
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.core.ParameterizedTypeReference;
+import org.springframework.core.log.LogAccessor;
 import org.springframework.util.Assert;
+import org.springframework.util.StringUtils;
 
 /**
  * The {@link AmqpTemplate} for RabbitMQ AMQP 1.0 protocol support.
@@ -55,6 +60,8 @@ import org.springframework.util.Assert;
  */
 public class RabbitAmqpTemplate implements AsyncAmqpTemplate, DisposableBean {
 
+	private static final LogAccessor LOG = new LogAccessor(RabbitAmqpAdmin.class);
+
 	private final AmqpConnectionFactory connectionFactory;
 
 	private final Lock instanceLock = new ReentrantLock();
@@ -71,10 +78,14 @@ public class RabbitAmqpTemplate implements AsyncAmqpTemplate, DisposableBean {
 
 	private @Nullable String defaultReceiveQueue;
 
+	private @Nullable String defaultReplyToQueue;
+
 	private Resource.StateListener @Nullable [] stateListeners;
 
 	private Duration publishTimeout = Duration.ofSeconds(60);
 
+	private Duration completionTimeout = Duration.ofSeconds(60);
+
 	public RabbitAmqpTemplate(AmqpConnectionFactory connectionFactory) {
 		this.connectionFactory = connectionFactory;
 	}
@@ -87,6 +98,19 @@ public class RabbitAmqpTemplate implements AsyncAmqpTemplate, DisposableBean {
 		this.publishTimeout = timeout;
 	}
 
+	/**
+	 * Set a duration for {@link CompletableFuture#orTimeout(long, TimeUnit)} on returns.
+	 * There is no {@link CompletableFuture} API like {@code onTimeout()} requested
+	 * from the {@link CompletableFuture#get(long, TimeUnit)},
+	 * but used in operations AMQP resources have to be closed eventually independently
+	 * of the {@link CompletableFuture} fulfilment.
+	 * Defaults to 1 minute.
+	 * @param completionTimeout duration for future completions.
+	 */
+	public void setCompletionTimeout(Duration completionTimeout) {
+		this.completionTimeout = completionTimeout;
+	}
+
 	/**
 	 * Set a default exchange for publishing.
 	 * Cannot be real default AMQP exchange.
@@ -116,6 +140,22 @@ public class RabbitAmqpTemplate implements AsyncAmqpTemplate, DisposableBean {
 		this.defaultQueue = queue;
 	}
 
+	/**
+	 * The name of the default queue to receive messages from when none is specified explicitly.
+	 * @param queue the default queue name to use for receive operation.
+	 */
+	public void setReceiveQueue(String queue) {
+		this.defaultReceiveQueue = queue;
+	}
+
+	/**
+	 * The name of the default queue to receive replies from when none is specified explicitly.
+	 * @param queue the default queue name to use for send-n-receive operation.
+	 */
+	public void setReplyToQueue(String queue) {
+		this.defaultReplyToQueue = queue;
+	}
+
 	/**
 	 * Set a converter for {@link #convertAndSend(Object)} operations.
 	 * @param messageConverter the converter.
@@ -124,14 +164,6 @@ public class RabbitAmqpTemplate implements AsyncAmqpTemplate, DisposableBean {
 		this.messageConverter = messageConverter;
 	}
 
-	/**
-	 * The name of the default queue to receive messages from when none is specified explicitly.
-	 * @param queue the default queue name to use for receive operation.
-	 */
-	public void setDefaultReceiveQueue(String queue) {
-		this.defaultReceiveQueue = queue;
-	}
-
 	private String getRequiredQueue() throws IllegalStateException {
 		String name = this.defaultReceiveQueue;
 		Assert.state(name != null, "No 'queue' specified. Check configuration of this 'RabbitAmqpTemplate'.");
@@ -178,7 +210,7 @@ public class RabbitAmqpTemplate implements AsyncAmqpTemplate, DisposableBean {
 	@Override
 	public CompletableFuture<Boolean> send(Message message) {
 		Assert.state(this.defaultExchange != null || this.defaultQueue != null,
-				"For send with defaults, an 'exchange' (and optional 'key') or 'queue' must be provided");
+				"For send with defaults, an 'exchange' (and optional 'routingKey') or 'queue' must be provided");
 		return doSend(this.defaultExchange, this.defaultRoutingKey, this.defaultQueue, message);
 	}
 
@@ -201,16 +233,8 @@ public class RabbitAmqpTemplate implements AsyncAmqpTemplate, DisposableBean {
 	private CompletableFuture<Boolean> doSend(@Nullable String exchange, @Nullable String routingKey,
 			@Nullable String queue, Message message) {
 
-		com.rabbitmq.client.amqp.Message amqpMessage = getPublisher().message();
-		com.rabbitmq.client.amqp.Message.MessageAddressBuilder address = amqpMessage.toAddress();
-		JavaUtils.INSTANCE
-				.acceptIfNotNull(exchange, address::exchange)
-				.acceptIfNotNull(routingKey, address::key)
-				.acceptIfNotNull(queue, address::queue);
-
-		amqpMessage = address.message();
-
-		RabbitAmqpUtils.toAmqpMessage(message, amqpMessage);
+		com.rabbitmq.client.amqp.Message amqpMessage =
+				toAmqpMessage(exchange, routingKey, queue, message, getPublisher()::message);
 
 		CompletableFuture<Boolean> publishResult = new CompletableFuture<>();
 
@@ -235,7 +259,7 @@ public class RabbitAmqpTemplate implements AsyncAmqpTemplate, DisposableBean {
 	@Override
 	public CompletableFuture<Boolean> convertAndSend(Object message) {
 		Assert.state(this.defaultExchange != null || this.defaultQueue != null,
-				"For send with defaults, an 'exchange' (and optional 'key') or 'queue' must be provided");
+				"For send with defaults, an 'exchange' (and optional 'routingKey') or 'queue' must be provided");
 		return doConvertAndSend(this.defaultExchange, this.defaultRoutingKey, this.defaultQueue, message, null);
 	}
 
@@ -273,10 +297,7 @@ public class RabbitAmqpTemplate implements AsyncAmqpTemplate, DisposableBean {
 	private CompletableFuture<Boolean> doConvertAndSend(@Nullable String exchange, @Nullable String routingKey,
 			@Nullable String queue, Object data, @Nullable MessagePostProcessor messagePostProcessor) {
 
-		Message message =
-				data instanceof Message
-						? (Message) data
-						: this.messageConverter.toMessage(data, new MessageProperties());
+		Message message = convertToMessageIfNecessary(data);
 		if (messagePostProcessor != null) {
 			message = messagePostProcessor.postProcessMessage(message);
 		}
@@ -288,6 +309,13 @@ public class RabbitAmqpTemplate implements AsyncAmqpTemplate, DisposableBean {
 		return receive(getRequiredQueue());
 	}
 
+	/**
+	 * Request a head message from the provided queue.
+	 * A returned {@link CompletableFuture} timeouts after {@link #setCompletionTimeout(Duration)}.
+	 * @param queueName the queue to consume message from.
+	 * @return the future with a received message.
+	 * @see #setCompletionTimeout(Duration)
+	 */
 	@SuppressWarnings("try")
 	@Override
 	public CompletableFuture<Message> receive(String queueName) {
@@ -306,47 +334,54 @@ public class RabbitAmqpTemplate implements AsyncAmqpTemplate, DisposableBean {
 						.build();
 
 		return messageFuture
-				.orTimeout(1, TimeUnit.MINUTES)
+				.orTimeout(this.completionTimeout.toMillis(), TimeUnit.MILLISECONDS)
 				.whenComplete((message, exception) -> consumer.close());
 	}
 
 	@Override
 	public CompletableFuture<Object> receiveAndConvert() {
-		return receiveAndConvert(getRequiredQueue());
+		return receiveAndConvert((ParameterizedTypeReference<Object>) null);
 	}
 
 	@Override
 	public CompletableFuture<Object> receiveAndConvert(String queueName) {
-		return receive(queueName)
-				.thenApply(this.messageConverter::fromMessage);
+		return receiveAndConvert(queueName, null);
 	}
 
 	/**
-	 * Receive a message from {@link #setDefaultReceiveQueue(String)} and convert its body
+	 * Receive a message from {@link #setReceiveQueue(String)} and convert its body
 	 * to the expected type.
 	 * The {@link #setMessageConverter(MessageConverter)} must be an implementation of {@link SmartMessageConverter}.
 	 * @param type the type to covert received result.
 	 * @return the CompletableFuture with a result.
 	 */
 	@Override
-	public <T> CompletableFuture<T> receiveAndConvert(ParameterizedTypeReference<T> type) {
+	public <T> CompletableFuture<T> receiveAndConvert(@Nullable ParameterizedTypeReference<T> type) {
 		return receiveAndConvert(getRequiredQueue(), type);
 	}
 
 	/**
-	 * Receive a message from {@link #setDefaultReceiveQueue(String)} and convert its body
+	 * Receive a message from {@link #setReceiveQueue(String)} and convert its body
 	 * to the expected type.
 	 * The {@link #setMessageConverter(MessageConverter)} must be an implementation of {@link SmartMessageConverter}.
 	 * @param queueName the queue to consume message from.
 	 * @param type the type to covert received result.
 	 * @return the CompletableFuture with a result.
 	 */
-	@SuppressWarnings("unchecked")
 	@Override
-	public <T> CompletableFuture<T> receiveAndConvert(String queueName, ParameterizedTypeReference<T> type) {
-		SmartMessageConverter smartMessageConverter = getRequiredSmartMessageConverter();
+	public <T> CompletableFuture<T> receiveAndConvert(String queueName, @Nullable ParameterizedTypeReference<T> type) {
 		return receive(queueName)
-				.thenApply((message) -> (T) smartMessageConverter.fromMessage(message, type));
+				.thenApply((message) -> convertReply(message, type));
+	}
+
+	@SuppressWarnings("unchecked")
+	private <T> T convertReply(Message message, @Nullable ParameterizedTypeReference<T> type) {
+		if (type != null) {
+			return (T) getRequiredSmartMessageConverter().fromMessage(message, type);
+		}
+		else {
+			return (T) this.messageConverter.fromMessage(message);
+		}
 	}
 
 	private SmartMessageConverter getRequiredSmartMessageConverter() throws IllegalStateException {
@@ -355,103 +390,280 @@ public class RabbitAmqpTemplate implements AsyncAmqpTemplate, DisposableBean {
 		return (SmartMessageConverter) this.messageConverter;
 	}
 
-	public <R, S> boolean receiveAndReply(ReceiveAndReplyCallback<R, S> callback) throws AmqpException {
-		throw new UnsupportedOperationException();
+	@Override
+	public <R, S> CompletableFuture<Boolean> receiveAndReply(ReceiveAndReplyCallback<R, S> callback) {
+		return receiveAndReply(getRequiredQueue(), callback);
 	}
 
-	public <R, S> boolean receiveAndReply(String queueName, ReceiveAndReplyCallback<R, S> callback) throws AmqpException {
-		throw new UnsupportedOperationException();
+	@Override
+	@SuppressWarnings("try")
+	public <R, S> CompletableFuture<Boolean> receiveAndReply(String queueName, ReceiveAndReplyCallback<R, S> callback) {
+		CompletableFuture<Boolean> rpcFuture = new CompletableFuture<>();
+
+		Consumer.MessageHandler consumerHandler =
+				(context, message) -> {
+					Message requestMessage = RabbitAmqpUtils.fromAmqpMessage(message, null);
+					try {
+						Object messageId = message.messageId();
+						Assert.notNull(messageId,
+								"The 'message-id' property has to be set on request. Used for reply correlation.");
+						String replyTo = message.replyTo();
+						Assert.hasText(replyTo,
+								"The 'reply-to' property has to be set on request. Used for reply publishing.");
+						Message reply = handleRequestAndProduceReply(requestMessage, callback);
+						if (reply == null) {
+							LOG.info(() -> "No reply for request: " + requestMessage);
+							context.accept();
+							rpcFuture.complete(false);
+						}
+						else {
+							com.rabbitmq.client.amqp.Message replyMessage = getPublisher().message();
+							RabbitAmqpUtils.toAmqpMessage(reply, replyMessage);
+							replyMessage.correlationId(messageId);
+							replyMessage.to(replyTo);
+							getPublisher().publish(replyMessage, (ctx) -> {
+							});
+							context.accept();
+							rpcFuture.complete(true);
+						}
+					}
+					catch (Exception ex) {
+						context.discard();
+						rpcFuture.completeExceptionally(
+								new AmqpIllegalStateException("Failed to process RPC request: " + requestMessage, ex));
+					}
+				};
+
+		Consumer consumer =
+				this.connectionFactory.getConnection()
+						.consumerBuilder()
+						.queue(queueName)
+						.initialCredits(1)
+						.priority(10)
+						.messageHandler(consumerHandler)
+						.build();
+
+		return rpcFuture
+				.orTimeout(this.completionTimeout.toMillis(), TimeUnit.MILLISECONDS)
+				.whenComplete((message, exception) -> consumer.close());
 	}
 
-	public <R, S> boolean receiveAndReply(ReceiveAndReplyCallback<R, S> callback, String replyExchange, String replyRoutingKey) throws AmqpException {
-		throw new UnsupportedOperationException();
+	@SuppressWarnings("unchecked")
+	private <S, R> @Nullable Message handleRequestAndProduceReply(Message requestMessage,
+			ReceiveAndReplyCallback<R, S> callback) {
+
+		Object receive = requestMessage;
+		if (!(ReceiveAndReplyMessageCallback.class.isAssignableFrom(callback.getClass()))) {
+			receive = this.messageConverter.fromMessage(requestMessage);
+		}
+
+		S reply;
+		try {
+			reply = callback.handle((R) receive);
+		}
+		catch (ClassCastException ex) {
+			StackTraceElement[] trace = ex.getStackTrace();
+			if (trace[0].getMethodName().equals("handle")
+					&& Objects.equals(trace[1].getFileName(), "RabbitAmqpTemplate.java")) {
+
+				throw new IllegalArgumentException("ReceiveAndReplyCallback '" + callback
+						+ "' can't handle received object '" + receive + "'", ex);
+			}
+			else {
+				throw ex;
+			}
+		}
+
+		if (reply != null) {
+			return convertToMessageIfNecessary(reply);
+		}
+		return null;
 	}
 
-	public <R, S> boolean receiveAndReply(String queueName, ReceiveAndReplyCallback<R, S> callback, String replyExchange, String replyRoutingKey) throws AmqpException {
-		throw new UnsupportedOperationException();
-	}
-
-	public <R, S> boolean receiveAndReply(ReceiveAndReplyCallback<R, S> callback, ReplyToAddressCallback<S> replyToAddressCallback) throws AmqpException {
-		throw new UnsupportedOperationException();
-	}
-
-	public <R, S> boolean receiveAndReply(String queueName, ReceiveAndReplyCallback<R, S> callback, ReplyToAddressCallback<S> replyToAddressCallback) throws AmqpException {
-		throw new UnsupportedOperationException();
+	private Message convertToMessageIfNecessary(Object data) {
+		if (data instanceof Message msg) {
+			return msg;
+		}
+		else {
+			return this.messageConverter.toMessage(data, new MessageProperties());
+		}
 	}
 
 	@Override
 	public CompletableFuture<Message> sendAndReceive(Message message) {
-		throw new UnsupportedOperationException();
+		Assert.state(this.defaultExchange != null || this.defaultQueue != null,
+				"For send-n-receive with defaults, an 'exchange' (and optional 'routingKey') or 'queue' must be provided");
+		return doSendAndReceive(this.defaultExchange, this.defaultRoutingKey, this.defaultQueue, message);
 	}
 
 	@Override
-	public CompletableFuture<Message> sendAndReceive(String routingKey, Message message) {
-		throw new UnsupportedOperationException();
+	public CompletableFuture<Message> sendAndReceive(String exchange, @Nullable String routingKey, Message message) {
+		return doSendAndReceive(exchange, routingKey != null ? routingKey : this.defaultRoutingKey, null, message);
 	}
 
 	@Override
-	public CompletableFuture<Message> sendAndReceive(String exchange, String routingKey, Message message) {
-		throw new UnsupportedOperationException();
+	public CompletableFuture<Message> sendAndReceive(String queue, Message message) {
+		return doSendAndReceive(null, null, queue, message);
+	}
+
+	@SuppressWarnings("try")
+	private CompletableFuture<Message> doSendAndReceive(@Nullable String exchange, @Nullable String routingKey,
+			@Nullable String queue, Message message) {
+
+		MessageProperties messageProperties = message.getMessageProperties();
+		String messageId = messageProperties.getMessageId();
+		String correlationId = messageProperties.getCorrelationId();
+		String replyTo = messageProperties.getReplyTo();
+
+		// HTTP over AMQP 1.0 extension specification, 5.1:
+		// To associate a response with a request, the correlation-id value of the response properties
+		// MUST be set to the message-id value of the request properties.
+		// So, this supplier will override request message-id, respectively.
+		// Otherwise, the RpcClient generates correlation-id internally.
+		Supplier<Object> correlationIdSupplier = null;
+		if (StringUtils.hasText(correlationId)) {
+			correlationIdSupplier = () -> correlationId;
+		}
+		else if (StringUtils.hasText(messageId)) {
+			correlationIdSupplier = () -> messageId;
+		}
+
+		// The default reply-to queue, or the one supplied in the message.
+		// Otherwise, the RpcClient generates one as exclusive and auto-delete.
+		String replyToQueue = this.defaultReplyToQueue;
+		if (StringUtils.hasText(replyTo)) {
+			replyToQueue = replyTo;
+		}
+
+		RpcClient rpcClient =
+				this.connectionFactory.getConnection()
+						.rpcClientBuilder()
+						.requestTimeout(this.publishTimeout)
+						.correlationIdSupplier(correlationIdSupplier)
+						.replyToQueue(replyToQueue)
+						.build();
+
+		com.rabbitmq.client.amqp.Message amqpMessage =
+				toAmqpMessage(exchange, routingKey, queue, message, rpcClient::message);
+
+		return rpcClient.publish(amqpMessage)
+				.thenApply((reply) -> RabbitAmqpUtils.fromAmqpMessage(reply, null))
+				.orTimeout(this.completionTimeout.toMillis(), TimeUnit.MILLISECONDS)
+				.whenComplete((replyMessage, exception) -> rpcClient.close());
 	}
 
 	@Override
 	public <C> CompletableFuture<C> convertSendAndReceive(Object object) {
-		throw new UnsupportedOperationException();
+		return convertSendAndReceiveAsType(object, null, null);
 	}
 
 	@Override
-	public <C> CompletableFuture<C> convertSendAndReceive(String routingKey, Object object) {
-		throw new UnsupportedOperationException();
+	public <C> CompletableFuture<C> convertSendAndReceive(String queue, Object object) {
+		return convertSendAndReceiveAsType(queue, object, null, null);
 	}
 
 	@Override
-	public <C> CompletableFuture<C> convertSendAndReceive(String exchange, String routingKey, Object object) {
-		throw new UnsupportedOperationException();
+	public <C> CompletableFuture<C> convertSendAndReceive(String exchange, @Nullable String routingKey, Object object) {
+		return convertSendAndReceiveAsType(exchange, routingKey, object, null, null);
 	}
 
 	@Override
 	public <C> CompletableFuture<C> convertSendAndReceive(Object object, MessagePostProcessor messagePostProcessor) {
-		throw new UnsupportedOperationException();
+		return convertSendAndReceiveAsType(object, messagePostProcessor, null);
 	}
 
 	@Override
-	public <C> CompletableFuture<C> convertSendAndReceive(String routingKey, Object object, MessagePostProcessor messagePostProcessor) {
-		throw new UnsupportedOperationException();
+	public <C> CompletableFuture<C> convertSendAndReceive(String queue, Object object,
+			MessagePostProcessor messagePostProcessor) {
+
+		return convertSendAndReceiveAsType(queue, object, messagePostProcessor, null);
 	}
 
 	@Override
-	public <C> CompletableFuture<C> convertSendAndReceive(String exchange, String routingKey, Object object, @Nullable MessagePostProcessor messagePostProcessor) {
-		throw new UnsupportedOperationException();
+	public <C> CompletableFuture<C> convertSendAndReceive(String exchange, @Nullable String routingKey,
+			Object object, @Nullable MessagePostProcessor messagePostProcessor) {
+
+		return convertSendAndReceiveAsType(exchange, routingKey, object, messagePostProcessor, null);
 	}
 
 	@Override
-	public <C> CompletableFuture<C> convertSendAndReceiveAsType(Object object, ParameterizedTypeReference<C> responseType) {
-		throw new UnsupportedOperationException();
+	public <C> CompletableFuture<C> convertSendAndReceiveAsType(Object object,
+			ParameterizedTypeReference<C> responseType) {
+
+		return convertSendAndReceiveAsType(object, null, responseType);
 	}
 
 	@Override
-	public <C> CompletableFuture<C> convertSendAndReceiveAsType(String routingKey, Object object, ParameterizedTypeReference<C> responseType) {
-		throw new UnsupportedOperationException();
+	public <C> CompletableFuture<C> convertSendAndReceiveAsType(String queue, Object object,
+			ParameterizedTypeReference<C> responseType) {
+
+		return convertSendAndReceiveAsType(queue, object, null, responseType);
 	}
 
 	@Override
-	public <C> CompletableFuture<C> convertSendAndReceiveAsType(String exchange, String routingKey, Object object, ParameterizedTypeReference<C> responseType) {
-		throw new UnsupportedOperationException();
+	public <C> CompletableFuture<C> convertSendAndReceiveAsType(String exchange, @Nullable String routingKey,
+			Object object, ParameterizedTypeReference<C> responseType) {
+
+		return convertSendAndReceiveAsType(exchange, routingKey, object, null, responseType);
 	}
 
 	@Override
-	public <C> CompletableFuture<C> convertSendAndReceiveAsType(Object object, MessagePostProcessor messagePostProcessor, ParameterizedTypeReference<C> responseType) {
-		throw new UnsupportedOperationException();
+	public <C> CompletableFuture<C> convertSendAndReceiveAsType(Object object,
+			@Nullable MessagePostProcessor messagePostProcessor, @Nullable ParameterizedTypeReference<C> responseType) {
+
+		Assert.state(this.defaultExchange != null || this.defaultQueue != null,
+				"For send with defaults, an 'exchange' (and optional 'routingKey') or 'queue' must be provided");
+
+		return doConvertSendAndReceive(this.defaultExchange, this.defaultRoutingKey, this.defaultQueue, object,
+				messagePostProcessor, responseType);
 	}
 
 	@Override
-	public <C> CompletableFuture<C> convertSendAndReceiveAsType(String routingKey, Object object, @Nullable MessagePostProcessor messagePostProcessor, @Nullable ParameterizedTypeReference<C> responseType) {
-		throw new UnsupportedOperationException();
+	public <C> CompletableFuture<C> convertSendAndReceiveAsType(String queue, Object object,
+			@Nullable MessagePostProcessor messagePostProcessor, @Nullable ParameterizedTypeReference<C> responseType) {
+
+		return doConvertSendAndReceive(null, null, queue, object, messagePostProcessor, responseType);
 	}
 
 	@Override
-	public <C> CompletableFuture<C> convertSendAndReceiveAsType(String exchange, String routingKey, Object object, @Nullable MessagePostProcessor messagePostProcessor, @Nullable ParameterizedTypeReference<C> responseType) {
-		throw new UnsupportedOperationException();
+	public <C> CompletableFuture<C> convertSendAndReceiveAsType(String exchange, @Nullable String routingKey,
+			Object object, @Nullable MessagePostProcessor messagePostProcessor,
+			@Nullable ParameterizedTypeReference<C> responseType) {
+
+		return doConvertSendAndReceive(exchange, routingKey != null ? routingKey : this.defaultRoutingKey, null,
+				object, messagePostProcessor, responseType);
+	}
+
+	private <C> CompletableFuture<C> doConvertSendAndReceive(@Nullable String exchange, @Nullable String routingKey,
+			@Nullable String queue, Object data, @Nullable MessagePostProcessor messagePostProcessor,
+			@Nullable ParameterizedTypeReference<C> responseType) {
+
+		Message message = convertToMessageIfNecessary(data);
+		if (messagePostProcessor != null) {
+			message = messagePostProcessor.postProcessMessage(message);
+		}
+		return doSendAndReceive(exchange, routingKey, queue, message)
+				.thenApply((reply) -> convertReply(reply, responseType));
+	}
+
+	private static com.rabbitmq.client.amqp.Message toAmqpMessage(@Nullable String exchange,
+			@Nullable String routingKey, @Nullable String queue, Message message,
+			Supplier<com.rabbitmq.client.amqp.Message> amqpMessageSupplier) {
+
+		com.rabbitmq.client.amqp.Message.MessageAddressBuilder address =
+				amqpMessageSupplier.get()
+						.toAddress();
+
+		JavaUtils.INSTANCE
+				.acceptIfNotNull(exchange, address::exchange)
+				.acceptIfNotNull(routingKey, address::key)
+				.acceptIfNotNull(queue, address::queue);
+
+		com.rabbitmq.client.amqp.Message amqpMessage = address.message();
+
+		RabbitAmqpUtils.toAmqpMessage(message, amqpMessage);
+
+		return amqpMessage;
 	}
 
 }

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

@@ -53,7 +53,8 @@ public final class RabbitAmqpUtils {
 				.acceptIfNotNull(amqpMessage.contentEncoding(), messageProperties::setContentEncoding)
 				.acceptIfNotNull(amqpMessage.absoluteExpiryTime(),
 						(exp) -> messageProperties.setExpiration(Long.toString(exp)))
-				.acceptIfNotNull(amqpMessage.creationTime(), (time) -> messageProperties.setTimestamp(new Date(time)));
+				.acceptIfNotNull(amqpMessage.creationTime(), (time) -> messageProperties.setTimestamp(new Date(time)))
+				.acceptIfNotNull(amqpMessage.replyTo(), messageProperties::setReplyTo);
 
 		amqpMessage.forEachProperty(messageProperties::setHeader);
 

spring-rabbitmq-client/src/test/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpTemplateTests.java

@@ -16,20 +16,31 @@
 
 package org.springframework.amqp.rabbitmq.client;
 
+import java.nio.charset.StandardCharsets;
 import java.time.Duration;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicReference;
 
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
+import org.springframework.amqp.AmqpIllegalStateException;
 import org.springframework.amqp.core.Binding;
 import org.springframework.amqp.core.BindingBuilder;
 import org.springframework.amqp.core.DirectExchange;
 import org.springframework.amqp.core.Message;
+import org.springframework.amqp.core.MessageBuilder;
 import org.springframework.amqp.core.Queue;
+import org.springframework.amqp.core.QueueBuilder;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.test.context.ContextConfiguration;
+import org.springframework.util.MimeTypeUtils;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatIllegalStateException;
@@ -58,11 +69,13 @@ public class RabbitAmqpTemplateTests extends RabbitAmqpTestBase {
 	void illegalStateOnNoDefaults() {
 		assertThatIllegalStateException()
 				.isThrownBy(() -> this.template.send(new Message(new byte[0])))
-				.withMessage("For send with defaults, an 'exchange' (and optional 'key') or 'queue' must be provided");
+				.withMessage(
+						"For send with defaults, an 'exchange' (and optional 'routingKey') or 'queue' must be provided");
 
 		assertThatIllegalStateException()
 				.isThrownBy(() -> this.template.convertAndSend(new byte[0]))
-				.withMessage("For send with defaults, an 'exchange' (and optional 'key') or 'queue' must be provided");
+				.withMessage(
+						"For send with defaults, an 'exchange' (and optional 'routingKey') or 'queue' must be provided");
 	}
 
 	@Test
@@ -81,7 +94,7 @@ public class RabbitAmqpTemplateTests extends RabbitAmqpTestBase {
 	@Test
 	void defaultQueues() {
 		this.rabbitAmqpTemplate.setQueue("q1");
-		this.rabbitAmqpTemplate.setDefaultReceiveQueue("q1");
+		this.rabbitAmqpTemplate.setReceiveQueue("q1");
 
 		assertThat(this.rabbitAmqpTemplate.convertAndSend("test2"))
 				.succeedsWithin(Duration.ofSeconds(10));
@@ -91,6 +104,65 @@ public class RabbitAmqpTemplateTests extends RabbitAmqpTestBase {
 				.isEqualTo("test2");
 	}
 
+	@Test
+	void verifyRpc() {
+		String testRequest = "rpc-request";
+		String testReply = "rpc-reply";
+
+		CompletableFuture<Object> rpcClientResult = this.template.convertSendAndReceive("e1", "k1", testRequest);
+
+		AtomicReference<String> receivedRequest = new AtomicReference<>();
+		CompletableFuture<Boolean> rpcServerResult =
+				this.rabbitAmqpTemplate.<String, String>receiveAndReply("q1",
+						payload -> {
+							receivedRequest.set(payload);
+							return testReply;
+						});
+
+		assertThat(rpcServerResult).succeedsWithin(Duration.ofSeconds(10)).isEqualTo(true);
+		assertThat(rpcClientResult).succeedsWithin(Duration.ofSeconds(10)).isEqualTo(testReply);
+		assertThat(receivedRequest.get()).isEqualTo(testRequest);
+
+		this.template.send("q1",
+				MessageBuilder.withBody("non-rpc-request".getBytes(StandardCharsets.UTF_8))
+						.setMessageId(UUID.randomUUID().toString())
+						.setContentType(MimeTypeUtils.TEXT_PLAIN_VALUE)
+						.build());
+
+		rpcServerResult = this.rabbitAmqpTemplate.<String, String>receiveAndReply("q1", payload -> "reply-attempt");
+
+		assertThat(rpcServerResult).failsWithin(Duration.ofSeconds(10))
+				.withThrowableOfType(ExecutionException.class)
+				.withCauseInstanceOf(AmqpIllegalStateException.class)
+				.withRootCauseInstanceOf(IllegalArgumentException.class)
+				.withMessageContaining("Failed to process RPC request: (Body:'non-rpc-request'")
+				.withStackTraceContaining("The 'reply-to' property has to be set on request. Used for reply publishing.");
+
+		rpcClientResult = this.template.convertSendAndReceive("q1", testRequest);
+		rpcServerResult = this.rabbitAmqpTemplate.<String, String>receiveAndReply("q1", payload -> null);
+
+		assertThat(rpcServerResult).succeedsWithin(Duration.ofSeconds(10)).isEqualTo(false);
+		assertThat(rpcClientResult).failsWithin(Duration.ofSeconds(2))
+				.withThrowableThat()
+				.isInstanceOf(TimeoutException.class);
+
+		this.template.convertSendAndReceive("q1", new byte[0]);
+
+		rpcServerResult = this.rabbitAmqpTemplate.<String, String>receiveAndReply("q1", payload -> payload);
+		assertThat(rpcServerResult).failsWithin(Duration.ofSeconds(10))
+				.withThrowableOfType(ExecutionException.class)
+				.withCauseInstanceOf(AmqpIllegalStateException.class)
+				.withRootCauseInstanceOf(ClassCastException.class)
+				.withMessageContaining("Failed to process RPC request: (Body:'[B")
+				.withStackTraceContaining("class [B cannot be cast to class java.lang.String");
+
+		assertThat(this.template.receiveAndConvert("dlq1")).succeedsWithin(10, TimeUnit.SECONDS)
+				.isEqualTo("non-rpc-request");
+
+		assertThat(this.template.receiveAndConvert("dlq1")).succeedsWithin(10, TimeUnit.SECONDS)
+				.isEqualTo(new byte[0]);
+	}
+
 	@Configuration
 	static class Config {
 
@@ -101,7 +173,7 @@ public class RabbitAmqpTemplateTests extends RabbitAmqpTestBase {
 
 		@Bean
 		Queue q1() {
-			return new Queue("q1");
+			return QueueBuilder.durable("q1").deadLetterExchange("dlx1").build();
 		}
 
 		@Bean

spring-rabbitmq-client/src/test/java/org/springframework/amqp/rabbitmq/client/RabbitAmqpTestBase.java

@@ -23,10 +23,13 @@ import java.util.stream.Stream;
 import com.rabbitmq.client.amqp.Environment;
 import com.rabbitmq.client.amqp.impl.AmqpEnvironmentBuilder;
 
+import org.springframework.amqp.core.Binding;
+import org.springframework.amqp.core.BindingBuilder;
 import org.springframework.amqp.core.Declarable;
 import org.springframework.amqp.core.Declarables;
 import org.springframework.amqp.core.Exchange;
 import org.springframework.amqp.core.Queue;
+import org.springframework.amqp.core.TopicExchange;
 import org.springframework.amqp.rabbit.junit.AbstractTestContainerTests;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.Lifecycle;
@@ -94,6 +97,21 @@ public abstract class RabbitAmqpTestBase extends AbstractTestContainerTests {
 			return new RabbitAmqpTemplate(connectionFactory);
 		}
 
+		@Bean
+		TopicExchange dlx1() {
+			return new TopicExchange("dlx1");
+		}
+
+		@Bean
+		Queue dlq1() {
+			return new Queue("dlq1");
+		}
+
+		@Bean
+		Binding dlq1Binding() {
+			return BindingBuilder.bind(dlq1()).to(dlx1()).with("#");
+		}
+
 		volatile boolean running;
 
 		@Override

spring-rabbitmq-client/src/test/java/org/springframework/amqp/rabbitmq/client/listener/RabbitAmqpListenerTests.java

@@ -30,11 +30,8 @@ import org.assertj.core.api.InstanceOfAssertFactories;
 import org.junit.jupiter.api.Test;
 
 import org.springframework.amqp.core.AmqpAcknowledgment;
-import org.springframework.amqp.core.Binding;
-import org.springframework.amqp.core.BindingBuilder;
 import org.springframework.amqp.core.Queue;
 import org.springframework.amqp.core.QueueBuilder;
-import org.springframework.amqp.core.TopicExchange;
 import org.springframework.amqp.rabbit.annotation.EnableRabbit;
 import org.springframework.amqp.rabbit.annotation.RabbitListener;
 import org.springframework.amqp.rabbit.annotation.RabbitListenerAnnotationBeanPostProcessor;
@@ -149,21 +146,6 @@ class RabbitAmqpListenerTests extends RabbitAmqpTestBase {
 	@EnableRabbit
 	static class Config {
 
-		@Bean
-		TopicExchange dlx1() {
-			return new TopicExchange("dlx1");
-		}
-
-		@Bean
-		Queue dlq1() {
-			return new Queue("dlq1");
-		}
-
-		@Bean
-		Binding dlq1Binding() {
-			return BindingBuilder.bind(dlq1()).to(dlx1()).with("#");
-		}
-
 		@Bean
 		Queue q1() {
 			return QueueBuilder.durable("q1").deadLetterExchange("dlx1").build();

spring-rabbitmq-client/src/test/resources/log4j2-test.xml

@@ -7,6 +7,7 @@
 	</Appenders>
 	<Loggers>
 		<Logger name="org.springframework.amqp.rabbit" level="info"/>
+		<Logger name="com.rabbitmq.client.amqp" level="warn"/>
 		<Root level="warn">
 			<AppenderRef ref="STDOUT" />
 		</Root>