feat: Receipt modack

google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/MessageDispatcher.java

@@ -89,6 +89,7 @@ class MessageDispatcher {
   private final LinkedBlockingQueue<AckRequestData> pendingAcks = new LinkedBlockingQueue<>();
   private final LinkedBlockingQueue<AckRequestData> pendingNacks = new LinkedBlockingQueue<>();
   private final LinkedBlockingQueue<AckRequestData> pendingReceipts = new LinkedBlockingQueue<>();
+  private final LinkedBlockingQueue<AckRequestData> exactlyOncePendingReceipts = new LinkedBlockingQueue<>();
 
   private final AtomicInteger messageDeadlineSeconds = new AtomicInteger();
   private final AtomicBoolean extendDeadline = new AtomicBoolean(true);
@@ -372,13 +373,46 @@ void processReceivedMessages(List<ReceivedMessage> messages) {
         // totally expire so that pubsub service sends us the message again.
         continue;
       }
-      outstandingBatch.add(new OutstandingMessage(message, ackHandler));
-      pendingReceipts.add(ackRequestData);
+      if (this.exactlyOnceDeliveryEnabled.get()) {
+        exactlyOncePendingReceipts.add(ackRequestData);
+      } else {
+        outstandingBatch.add(new OutstandingMessage(message, ackHandler));
+        pendingReceipts.add(ackRequestData);
+      }
+    }
+    if (this.exactlyOnceDeliveryEnabled.get()) {
+      List<ModackRequestData> modackRequestData = new ArrayList<ModackRequestData>();
+      List<AckRequestData> ackRequestDataReceipts = new ArrayList<AckRequestData>();
+      exactlyOncePendingReceipts.drainTo(ackRequestDataReceipts);
+      if (!ackRequestDataReceipts.isEmpty()) {
+        modackRequestData.add(
+            new ModackRequestData(this.getMessageDeadlineSeconds(), ackRequestDataReceipts));
+      }
+      ackProcessor.sendModackOperations(modackRequestData);
+      ApiFuture.addCallback(ackReqData.messageFuture, new ApiFutureCallback<string>() {
+        @Override
+        public void onFailure(Throwable throwable) {
+          System.out.println("Error with receipt modack");
+        }
+        @Override
+        public void onSuccess() {
+          // outstandingBatch.add(new OutstandingMessage(message, ackHandler));
+          // processBatch(outstandingBatch);
+          try {
+            flowController.reserve(1, message.receivedMessage.getMessage().getSerializedSize());
+          } catch (FlowControlException unexpectedException) {
+            // This should be a blocking flow controller and never throw an exception.
+            throw new IllegalStateException("Flow control unexpected exception", unexpectedException);
+          }
+          processOutstandingMessage(addDeliveryInfoCount(message.receivedMessage), message.ackHandler);
+        }
+      }, MoreExecutors.directExecutor());
+    } else {
+      processBatch(outstandingBatch);
     }
-
-    processBatch(outstandingBatch);
   }
 
+
   private void processBatch(List<OutstandingMessage> batch) {
     messagesWaiter.incrementPendingCount(batch.size());
     for (OutstandingMessage message : batch) {