Wait for openFuture on AMQP links before usage

Author: artembilan

spring-amqp-client/src/main/java/org/springframework/amqp/client/DefaultAmqpClient.java

@@ -160,7 +160,8 @@ private Sender getSender() throws ClientException {
 							this.connectionFactory
 									.getConnection()
 									.openAnonymousSender(this.senderOptions);
-					this.sender = senderToReturn;
+					Future<Sender> openFuture = senderToReturn.openFuture();
+					this.sender = ProtonUtils.toSupplier(openFuture, this.senderOptions.openTimeout()).get();
 				}
 			}
 			finally {
@@ -172,9 +173,12 @@ private Sender getSender() throws ClientException {
 
 	private <M extends Message<?>> CompletableFuture<M> receive(String fromAddress, @Nullable Duration receiveTimeout) {
 		try {
-			Receiver receiver =
+			Future<Receiver> openFuture =
 					this.connectionFactory.getConnection()
 							.openReceiver(fromAddress, this.receiverOptions)
+							.openFuture();
+			Receiver receiver =
+					ProtonUtils.toSupplier(openFuture, this.receiverOptions.openTimeout()).get()
 							// Since this 'Receiver' is volatile and only about one message to consume,
 							// therefore only one credit is permitted without renewing.
 							.addCredit(1);

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

@@ -23,6 +23,7 @@
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.Executor;
+import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
@@ -283,8 +284,13 @@ public void start() {
 				for (String queue : this.queues) {
 					for (int i = 0; i < this.consumersPerQueue; i++) {
 						try {
+							Future<Receiver> openFuture =
+									this.connectionFactory.getConnection()
+											.openReceiver(queue, receiverOptions)
+											.openFuture();
 							ClientReceiver receiver =
-									(ClientReceiver) connection.openReceiver(queue, receiverOptions)
+									(ClientReceiver) ProtonUtils.toSupplier(openFuture, receiverOptions.openTimeout())
+											.get()
 											.addCredit(this.initialCredits);
 							AmqpConsumer consumer = new AmqpConsumer(receiver);
 							this.queueToConsumers.add(queue, consumer);

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

@@ -21,9 +21,12 @@
 import java.lang.reflect.WildcardType;
 import java.util.Objects;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Future;
 
+import org.apache.qpid.protonj2.client.ConnectionOptions;
 import org.apache.qpid.protonj2.client.Delivery;
 import org.apache.qpid.protonj2.client.DeliveryState;
+import org.apache.qpid.protonj2.client.Sender;
 import org.apache.qpid.protonj2.client.Tracker;
 import org.apache.qpid.protonj2.client.exceptions.ClientException;
 import org.jspecify.annotations.Nullable;
@@ -566,10 +569,14 @@ private static void settleRequest(Delivery delivery, Message request, AmqpAcknow
 
 	private static void sendResponse(Delivery delivery, String replyTo, Message response) throws Exception {
 		org.apache.qpid.protonj2.client.Message<?> protonMessage = ProtonUtils.toProtonMessage(response);
-		Tracker tracker =
+		Future<Sender> openFuture =
 				delivery.receiver()
 						.connection()
 						.openSender(replyTo)
+						.openFuture();
+		Tracker tracker =
+				ProtonUtils.toSupplier(openFuture, ConnectionOptions.DEFAULT_OPEN_TIMEOUT)
+						.get()
 						.send(protonMessage)
 						.settlementFuture()
 						.get();