---

dpcollins-google · sduskis · commit 7058ec011c49 · 2019-03-04T08:38:01.000-05:00
yaml --- r: 18935 b: refs/heads/autosynth-vision c: 77fb0ca h: refs/heads/master i: 18933: 5832589 18931: 2b2cc5d 18927: a5d646f
diff --git a/[refs] b/[refs]
@@ -117,7 +117,7 @@ refs/heads/autosynth-container: f6384095f50b31bfc8208542a49181e158d3681c
 refs/heads/autosynth-monitoring: 832528c131ac32b115fd7a369701d9e6e4ec62e5
 refs/heads/autosynth-pubsub: 7a327eaed5fb2a5d6c1691feb73c3e0194dd805e
 refs/heads/autosynth-video-intelligence: 53ca56e5d077bf67065b730f3fe621f40689fb4c
-refs/heads/autosynth-vision: aa5f27188cc55ebb5773b6597c6c49c965a40add
+refs/heads/autosynth-vision: 77fb0ca8e507c38b4b559836b5c55576ef50e0c9
 refs/heads/spanner: b01127f885b4611bf1852abb0ce481eeb7fcc131
 refs/tags/v0.68.0: 9cc799fcf68c82ab431d425fefa58ef615ce8e5b
 refs/tags/v0.69.0: 78f67a29e8b9c46ba01de566a2eae0fd1c03edea
diff --git a/branches/autosynth-vision/google-cloud-clients/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/MessageDispatcher.java b/branches/autosynth-vision/google-cloud-clients/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/MessageDispatcher.java
@@ -402,51 +402,53 @@ public void processOutstandingBatches() {
           batchCallback = nextBatch.doneCallback;
         }
       }
-
-      final PubsubMessage message = outstandingMessage.receivedMessage().getMessage();
-      final AckHandler ackHandler = outstandingMessage.ackHandler();
-      final SettableApiFuture<AckReply> response = SettableApiFuture.create();
-      final AckReplyConsumer consumer =
-          new AckReplyConsumer() {
-            @Override
-            public void ack() {
-              response.set(AckReply.ACK);
-            }
-
-            @Override
-            public void nack() {
-              response.set(AckReply.NACK);
-            }
-          };
-      ApiFutures.addCallback(response, ackHandler, MoreExecutors.directExecutor());
-      executor.execute(
-          new Runnable() {
-            @Override
-            public void run() {
-              try {
-                if (ackHandler
-                    .totalExpiration
-                    .plusSeconds(messageDeadlineSeconds.get())
-                    .isBefore(now())) {
-                  // Message expired while waiting. We don't extend these messages anymore,
-                  // so it was probably sent to someone else. Don't work on it.
-                  // Don't nack it either, because we'd be nacking someone else's message.
-                  ackHandler.forget();
-                  return;
-                }
-
-                receiver.receiveMessage(message, consumer);
-              } catch (Exception e) {
-                response.setException(e);
-              }
-            }
-          });
+      processOutstandingMessage(
+          outstandingMessage.receivedMessage.getMessage(), outstandingMessage.ackHandler);
       if (batchDone) {
         batchCallback.run();
       }
     }
   }
 
+  private void processOutstandingMessage(final PubsubMessage message, final AckHandler ackHandler) {
+    final SettableApiFuture<AckReply> response = SettableApiFuture.create();
+    final AckReplyConsumer consumer =
+        new AckReplyConsumer() {
+          @Override
+          public void ack() {
+            response.set(AckReply.ACK);
+          }
+
+          @Override
+          public void nack() {
+            response.set(AckReply.NACK);
+          }
+        };
+    ApiFutures.addCallback(response, ackHandler, MoreExecutors.directExecutor());
+    executor.execute(
+        new Runnable() {
+          @Override
+          public void run() {
+            try {
+              if (ackHandler
+                  .totalExpiration
+                  .plusSeconds(messageDeadlineSeconds.get())
+                  .isBefore(now())) {
+                // Message expired while waiting. We don't extend these messages anymore,
+                // so it was probably sent to someone else. Don't work on it.
+                // Don't nack it either, because we'd be nacking someone else's message.
+                ackHandler.forget();
+                return;
+              }
+
+              receiver.receiveMessage(message, consumer);
+            } catch (Exception e) {
+              response.setException(e);
+            }
+          }
+        });
+  }
+
   /** Compute the ideal deadline, set subsequent modacks to this deadline, and return it. */
   @InternalApi
   int computeDeadlineSeconds() {
