Use a CompletableFuture for publishing settlement callback

Instead of a Future. There is no need to poll the settlement future, the
callback is called asynchronously from the executor task. This improved
the throughput by about 75%.

Author: acogoluegnes

src/main/java/com/rabbitmq/client/amqp/impl/AmqpPublisher.java

@@ -107,20 +107,22 @@ public void publish(Message message, Callback callback) {
             message,
             this.connectionInfo,
             publishCall);
-    this.executorService.submit(
-        () -> {
-          Status status;
-          try {
-            // FIXME set a timeout for publishing settlement
-            tracker.settlementFuture().get();
-            status = mapDeliveryState(tracker.remoteState());
-          } catch (InterruptedException | ExecutionException e) {
-            status = Status.REJECTED;
-          }
-          DefaultContext defaultContext = new DefaultContext(message, status);
-          this.metricsCollector.publishDisposition(mapToPublishDisposition(status));
-          callback.handle(defaultContext);
-        });
+    tracker
+        .settlementFuture()
+        .handleAsync(
+            (t, ex) -> {
+              Status status;
+              if (ex == null) {
+                status = mapDeliveryState(t.remoteState());
+              } else {
+                status = Status.REJECTED;
+              }
+              DefaultContext defaultContext = new DefaultContext(message, status);
+              this.metricsCollector.publishDisposition(mapToPublishDisposition(status));
+              callback.handle(defaultContext);
+              return null;
+            },
+            this.executorService);
     this.metricsCollector.publish();
   }
 

src/main/qpid/org/apache/qpid/protonj2/client/Tracker.java

@@ -16,6 +16,7 @@
  */
 package org.apache.qpid.protonj2.client;
 
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 
@@ -90,7 +91,7 @@ public interface Tracker {
      *
      * @return a {@link Future} that can be used to wait on remote settlement.
      */
-    Future<Tracker> settlementFuture();
+    CompletableFuture<Tracker> settlementFuture();
 
     /**
      * Waits if necessary for the remote to settle the sent delivery unless it has

src/main/qpid/org/apache/qpid/protonj2/client/impl/ClientNoOpTracker.java

@@ -16,6 +16,7 @@
  */
 package org.apache.qpid.protonj2.client.impl;
 
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 
@@ -80,8 +81,10 @@ public boolean remoteSettled() {
     }
 
     @Override
-    public Future<Tracker> settlementFuture() {
-        return ClientFutureFactory.completedFuture(this);
+    public CompletableFuture<Tracker> settlementFuture() {
+        CompletableFuture<Tracker> result = new CompletableFuture<>();
+        result.complete(this);
+        return result;
     }
 
     @Override

src/main/qpid/org/apache/qpid/protonj2/client/impl/ClientSender.java

@@ -214,7 +214,7 @@ private void failPendingUnsettledAndBlockedSends(ClientException cause) {
         // Cancel all settlement futures for in-flight sends passing an appropriate error to the future
         protonSender.unsettled().forEach((delivery) -> {
             try {
-                delivery.getLinkedResource(ClientTrackable.class).settlementFuture().failed(cause);
+                delivery.getLinkedResource(ClientTrackable.class).settlementFuture().completeExceptionally(cause);
             } catch (Exception e) {
             }
         });

src/main/qpid/org/apache/qpid/protonj2/client/impl/ClientStreamSender.java

@@ -358,7 +358,7 @@ private void failPendingUnsettledAndBlockedSends(ClientException cause) {
         // Cancel all settlement futures for in-flight sends passing an appropriate error to the future
         protonSender.unsettled().forEach((delivery) -> {
             try {
-                delivery.getLinkedResource(ClientTrackable.class).settlementFuture().failed(cause);
+                delivery.getLinkedResource(ClientTrackable.class).settlementFuture().completeExceptionally(cause);
             } catch (Exception e) {
             }
         });

src/main/qpid/org/apache/qpid/protonj2/client/impl/ClientTrackable.java

@@ -18,6 +18,7 @@
 package org.apache.qpid.protonj2.client.impl;
 
 import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
@@ -49,7 +50,7 @@ public abstract class ClientTrackable<SenderType extends ClientSenderLinkType<?>
     protected static final AtomicReferenceFieldUpdater<ClientTrackable, DeliveryState> REMOTEL_DELIVERY_STATE_UPDATER =
         AtomicReferenceFieldUpdater.newUpdater(ClientTrackable.class, DeliveryState.class, "remoteDeliveryState");
 
-    private ClientFuture<TrackerType> remoteSettlementFuture;
+    private final CompletableFuture<TrackerType> remoteSettlementFuture;
     private volatile int remotelySettled;
     private volatile DeliveryState remoteDeliveryState;
 
@@ -64,6 +65,7 @@ public abstract class ClientTrackable<SenderType extends ClientSenderLinkType<?>
     ClientTrackable(SenderType sender, OutgoingDelivery delivery) {
         Objects.requireNonNull(sender, "Sender cannot be null for a Tracker");
 
+        this.remoteSettlementFuture = new CompletableFuture<>();
         this.sender = sender;
         this.delivery = delivery;
         this.delivery.deliveryStateUpdatedHandler(this::processDeliveryUpdated);
@@ -92,12 +94,7 @@ public TrackerType disposition(DeliveryState state, boolean settle) throws Clien
             sender.disposition(delivery, ClientDeliveryState.asProtonType(state), settle);
         } finally {
             if (settle) {
-                synchronized (this) {
-                    if (remoteSettlementFuture == null) {
-                        remoteSettlementFuture = sender.session.connection().getFutureFactory().createFuture();
-                    }
-                    remoteSettlementFuture.complete(self());
-                }
+                remoteSettlementFuture.complete(self());
             }
         }
 
@@ -108,12 +105,7 @@ public TrackerType settle() throws ClientException {
         try {
             sender.disposition(delivery, null, true);
         } finally {
-            synchronized (this) {
-                if (remoteSettlementFuture == null) {
-                    remoteSettlementFuture = sender.session.connection().getFutureFactory().createFuture();
-                }
-                remoteSettlementFuture.complete(self());
-            }
+            remoteSettlementFuture.complete(self());
         }
 
         return self();
@@ -123,17 +115,10 @@ public synchronized boolean settled() {
         return delivery.isSettled();
     }
 
-    public ClientFuture<TrackerType> settlementFuture() {
-        synchronized (this) {
-            if (remoteSettlementFuture == null) {
-                remoteSettlementFuture = sender.session.connection().getFutureFactory().createFuture();
-            }
-
-            if (delivery.isSettled() || remoteSettled()) {
-                remoteSettlementFuture.complete(self());
-            }
+    public CompletableFuture<TrackerType> settlementFuture() {
+        if (delivery.isSettled() || remoteSettled()) {
+            remoteSettlementFuture.complete(self());
         }
-
         return remoteSettlementFuture;
     }