Fix spring-rabbitmq-client for latest API

* Fix deprecation in the `RabbitConnectionFactoryBean`

Author: artembilan

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RabbitConnectionFactoryBean.java

@@ -59,7 +59,7 @@
 import org.springframework.util.StringUtils;
 
 /**
- * Factory bean to create a RabbitMQ ConnectionFactory, delegating most setter methods and
+ * The factory bean to create a RabbitMQ ConnectionFactory, delegating most setter methods and
  * optionally enabling SSL, with or without certificate validation. When
  * {@link #setSslPropertiesLocation(Resource) sslPropertiesLocation} is not null, the
  * default implementation loads a {@code PKCS12} keystore and a {@code JKS} truststore
@@ -80,6 +80,7 @@
  * @author Dominique Villard
  * @author Zachary DeLuca
  * @author Ngoc Nhan
+ * @author Artem Bilan
  *
  * @since 1.4
  */
@@ -167,16 +168,16 @@ public RabbitConnectionFactoryBean() {
 	}
 
 	/**
-	 * Whether Server Side certificate has to be validated or not.
-	 * @return true if Server Side certificate has to be skipped
+	 * Whether a Server Side certificate has to be validated or not.
+	 * @return true if the Server Side certificate has to be skipped
 	 * @since 1.6.6
 	 */
 	public boolean isSkipServerCertificateValidation() {
 		return this.skipServerCertificateValidation;
 	}
 
 	/**
-	 * Whether Server Side certificate has to be validated or not.
+	 * Whether a Server Side certificate has to be validated or not.
 	 * This would be used if useSSL is set to true and should only be used on dev or Qa regions
 	 * skipServerCertificateValidation should <b> never be set to true in production</b>
 	 * @param skipServerCertificateValidation Flag to override Server side certificate checks;
@@ -598,8 +599,10 @@ public void setExceptionHandler(ExceptionHandler exceptionHandler) {
 	/**
 	 * Whether the factory should be configured to use Java NIO.
 	 * @param useNio true to use Java NIO, false to use blocking IO
+	 * @deprecated since 4.0 in favor of {@link #setUseNetty(boolean)}
 	 * @see com.rabbitmq.client.ConnectionFactory#useNio()
 	 */
+	@Deprecated(since = "4.0", forRemoval = true)
 	public void setUseNio(boolean useNio) {
 		if (useNio) {
 			this.connectionFactory.useNio();
@@ -609,10 +612,26 @@ public void setUseNio(boolean useNio) {
 		}
 	}
 
+	/**
+	 * Whether the factory should be configured to use Netty.
+	 * @param useNetty true to use Netty, false to use blocking IO
+	 * @see com.rabbitmq.client.ConnectionFactory#netty()
+	 */
+	public void setUseNetty(boolean useNetty) {
+		if (useNetty) {
+			this.connectionFactory.netty();
+		}
+		else {
+			this.connectionFactory.useBlockingIo();
+		}
+	}
+
 	/**
 	 * @param nioParams the NIO parameters
+	 * @deprecated since 4.0 in favor of {@link #setUseNetty(boolean)}
 	 * @see com.rabbitmq.client.ConnectionFactory#setNioParams(com.rabbitmq.client.impl.nio.NioParams)
 	 */
+	@Deprecated(since = "4.0", forRemoval = true)
 	public void setNioParams(NioParams nioParams) {
 		this.connectionFactory.setNioParams(nioParams);
 	}
@@ -627,7 +646,7 @@ public void setMetricsCollector(MetricsCollector metricsCollector) {
 
 	/**
 	 * Set to true to enable amqp-client automatic recovery. Note: Spring AMQP
-	 * implements its own connection recovery and this is generally not needed.
+	 * implements its own connection recovery, and this is generally not needed.
 	 * @param automaticRecoveryEnabled true to enable.
 	 * @since 1.7.1
 	 */
@@ -638,7 +657,7 @@ public void setAutomaticRecoveryEnabled(boolean automaticRecoveryEnabled) {
 	/**
 	 * Set to true to enable amqp-client topology recovery. Note: if there is a
 	 * Rabbit admin in the application context, Spring AMQP
-	 * implements its own recovery and this is generally not needed.
+	 * implements its own recovery, and this is generally not needed.
 	 * @param topologyRecoveryEnabled true to enable.
 	 * @since 1.7.1
 	 */

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

@@ -28,8 +28,8 @@
 import com.rabbitmq.client.amqp.Consumer;
 import com.rabbitmq.client.amqp.Environment;
 import com.rabbitmq.client.amqp.Publisher;
+import com.rabbitmq.client.amqp.Requester;
 import com.rabbitmq.client.amqp.Resource;
-import com.rabbitmq.client.amqp.RpcClient;
 import org.jspecify.annotations.Nullable;
 
 import org.springframework.amqp.AmqpIllegalStateException;
@@ -171,7 +171,11 @@ private String getRequiredQueue() throws IllegalStateException {
 		return name;
 	}
 
-	private Publisher getPublisher() {
+	/**
+	 * Return the {@link Publisher} for low-level AMQP operations.
+	 * @return the {@link Publisher} for low-level AMQP operations.
+	 */
+	public Publisher getPublisher() {
 		Object publisherToReturn = this.publisher;
 		if (publisherToReturn == null) {
 			this.instanceLock.lock();
@@ -524,7 +528,7 @@ private CompletableFuture<Message> doSendAndReceive(@Nullable String exchange, @
 		// 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.
+		// Otherwise, the Requester generates correlation-id internally.
 		Supplier<Object> correlationIdSupplier = null;
 		if (StringUtils.hasText(correlationId)) {
 			correlationIdSupplier = () -> correlationId;
@@ -534,15 +538,15 @@ else if (StringUtils.hasText(messageId)) {
 		}
 
 		// The default reply-to queue, or the one supplied in the message.
-		// Otherwise, the RpcClient generates one as exclusive and auto-delete.
+		// Otherwise, the Requester generates one as exclusive and auto-delete.
 		String replyToQueue = this.defaultReplyToQueue;
 		if (StringUtils.hasText(replyTo)) {
 			replyToQueue = replyTo;
 		}
 
-		RpcClient rpcClient =
+		Requester rpcClient =
 				this.connectionFactory.getConnection()
-						.rpcClientBuilder()
+						.requesterBuilder()
 						.requestTimeout(this.publishTimeout)
 						.correlationIdSupplier(correlationIdSupplier)
 						.replyToQueue(replyToQueue)

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

@@ -24,12 +24,14 @@
 
 import com.rabbitmq.client.Channel;
 import com.rabbitmq.client.amqp.Consumer;
+import com.rabbitmq.client.amqp.Publisher;
 import org.jspecify.annotations.Nullable;
 
 import org.springframework.amqp.core.Address;
 import org.springframework.amqp.core.AmqpAcknowledgment;
 import org.springframework.amqp.core.Message;
 import org.springframework.amqp.core.MessagePostProcessor;
+import org.springframework.amqp.rabbit.core.AmqpNackReceivedException;
 import org.springframework.amqp.rabbit.listener.adapter.InvocationResult;
 import org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter;
 import org.springframework.amqp.rabbit.listener.api.RabbitListenerErrorHandler;
@@ -156,7 +158,18 @@ protected void sendResponse(@Nullable Channel channel, Address replyTo, Message
 		}
 		else {
 			Assert.hasText(replyToRoutingKey, "The 'replyTo' must be provided, in request message or in @SendTo.");
-			sendFuture = this.rabbitAmqpTemplate.send(replyToRoutingKey.replaceFirst("queues/", ""), replyMessage);
+			Publisher publisher = this.rabbitAmqpTemplate.getPublisher();
+			com.rabbitmq.client.amqp.Message amqpMessage = publisher.message();
+			RabbitAmqpUtils.toAmqpMessage(replyMessage, amqpMessage);
+			amqpMessage.to("/queues/" + replyToRoutingKey.replaceFirst("queues/", ""));
+			sendFuture = new CompletableFuture<>();
+			publisher.publish(amqpMessage, (context) -> {
+				switch (context.status()) {
+					case ACCEPTED -> sendFuture.complete(true);
+					case REJECTED, RELEASED -> sendFuture.completeExceptionally(
+							new AmqpNackReceivedException("The message was rejected", messageIn));
+				}
+			});
 		}
 
 		sendFuture.join();