---

dpcollins-google · sduskis · commit 3bc06759e595 · 2019-03-04T08:38:01.000-05:00
yaml --- r: 20647 b: refs/heads/autosynth-websecurityscanner c: 77fb0ca h: refs/heads/master i: 20645: fae09a8 20643: 199f230 20639: 76dda9d
diff --git a/[refs] b/[refs]
@@ -143,7 +143,7 @@ refs/heads/autosynth-speech: 75d6c62a9d07d3a3642980502a25d07fbde0f232
 refs/heads/autosynth-tasks: 5d6a4f686ffef2080506d3b613fc7563a8bba1cb
 refs/heads/autosynth-texttospeech: 2dcc5dc22be0f456caa1b6a8a4bcdace2641239c
 refs/heads/autosynth-trace: 8804c46bfe147702ee9c95669f17f42d3790cf23
-refs/heads/autosynth-websecurityscanner: aa5f27188cc55ebb5773b6597c6c49c965a40add
+refs/heads/autosynth-websecurityscanner: 77fb0ca8e507c38b4b559836b5c55576ef50e0c9
 refs/heads/bigquerystorage: 06db74d123d7f8a3ef48755c2fcabed09faf8e64
 refs/heads/elharo-patch-1: ce159ef828d3c545991ff78e7b6e0d912a9453e9
 refs/heads/mrschmidt-numericadd: 96509abadf6190b7886d57fdd9b090da55f5171c
diff --git a/branches/autosynth-websecurityscanner/google-cloud-clients/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/MessageDispatcher.java b/branches/autosynth-websecurityscanner/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() {
