---

dpcollins-google · sduskis · commit 3e4051d6b86e · 2019-03-04T10:50:40.000-05:00
yaml --- r: 19503 b: refs/heads/autosynth-dialogflow c: 5e0acc0 h: refs/heads/master i: 19501: 21c18cc 19499: caec36e 19495: da43354 19487: 947f9ed
diff --git a/[refs] b/[refs]
@@ -129,7 +129,7 @@ refs/heads/autosynth-bigtable: 45891a7178142c84a16d6a0b792f8ecb950a3159
 refs/heads/autosynth-bigtable-admin: 6379a2bc712f2736c83de0e009b4d26da4fa82ca
 refs/heads/autosynth-containeranalysis: 0f377286f60df2b6fd37c183bf170c68dc014cfb
 refs/heads/autosynth-datastore: 9acd400b484d6691a080c9152a331d88d24fefc1
-refs/heads/autosynth-dialogflow: 4517817a110e3d5230f63ccff810c0644cfab78a
+refs/heads/autosynth-dialogflow: 5e0acc0913b8608dd896a9e8da1241b8bc251a1d
 refs/heads/autosynth-errorreporting: 9891e73a56af7c097829ca7a521b0e862ba6af30
 refs/heads/autosynth-firestore: ddb29f8ee445a938fc40be8227dea87b3f508ab3
 refs/heads/autosynth-iot: f16fc4ba85bbb37eacecb64a2a7336f685813f6c
diff --git a/branches/autosynth-dialogflow/google-cloud-clients/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/MessageDispatcher.java b/branches/autosynth-dialogflow/google-cloud-clients/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/MessageDispatcher.java
@@ -371,41 +371,31 @@ public void processReceivedMessages(List<ReceivedMessage> messages, Runnable don
     processOutstandingBatches();
   }
 
-  public void processOutstandingBatches() {
-    while (true) {
-      boolean batchDone = false;
-      Runnable batchCallback = null;
-      OutstandingMessage outstandingMessage;
-      synchronized (outstandingMessageBatches) {
-        OutstandingMessageBatch nextBatch = outstandingMessageBatches.peek();
-        if (nextBatch == null) {
-          return;
-        }
-        outstandingMessage = nextBatch.messages.peek();
-        if (outstandingMessage == null) {
-          return;
-        }
-        try {
-          // This is a non-blocking flow controller.
-          flowController.reserve(
-              1, outstandingMessage.receivedMessage().getMessage().getSerializedSize());
-        } catch (FlowController.MaxOutstandingElementCountReachedException
-            | FlowController.MaxOutstandingRequestBytesReachedException flowControlException) {
-          return;
-        } catch (FlowControlException unexpectedException) {
-          throw new IllegalStateException("Flow control unexpected exception", unexpectedException);
-        }
-        nextBatch.messages.poll(); // We got a hold to the message already.
-        batchDone = nextBatch.messages.isEmpty();
-        if (batchDone) {
-          outstandingMessageBatches.poll();
-          batchCallback = nextBatch.doneCallback;
+  private void processOutstandingBatches() {
+    synchronized (outstandingMessageBatches) {
+      for (OutstandingMessageBatch nextBatch = outstandingMessageBatches.poll();
+          nextBatch != null;
+          nextBatch = outstandingMessageBatches.poll()) {
+        for (OutstandingMessage nextMessage = nextBatch.messages.poll();
+            nextMessage != null;
+            nextMessage = nextBatch.messages.poll()) {
+          try {
+            // This is a non-blocking flow controller.
+            flowController.reserve(1, nextMessage.receivedMessage.getMessage().getSerializedSize());
+          } catch (FlowController.MaxOutstandingElementCountReachedException
+              | FlowController.MaxOutstandingRequestBytesReachedException flowControlException) {
+            // Unwind previous changes in the batches outstanding.
+            nextBatch.messages.addFirst(nextMessage);
+            outstandingMessageBatches.addFirst(nextBatch);
+            return;
+          } catch (FlowControlException unexpectedException) {
+            throw new IllegalStateException(
+                "Flow control unexpected exception", unexpectedException);
+          }
+          processOutstandingMessage(
+              nextMessage.receivedMessage.getMessage(), nextMessage.ackHandler);
         }
-      }
-      processOutstandingMessage(
-          outstandingMessage.receivedMessage.getMessage(), outstandingMessage.ackHandler);
-      if (batchDone) {
-        batchCallback.run();
+        nextBatch.doneCallback.run();
       }
     }
   }
