---

dpcollins-google · sduskis · commit a69aacb6fbe7 · 2019-03-04T08:38:01.000-05:00
yaml --- r: 23239 b: refs/heads/autosynth-datastore c: 77fb0ca h: refs/heads/master i: 23237: b1e0809 23235: 281cadb 23231: 4a94a0e
diff --git a/[refs] b/[refs]
@@ -127,7 +127,7 @@ refs/heads/autosynth-bigquerystorage: d2c53da3b012e38c662e4df0738042435f19365f
 refs/heads/autosynth-bigtable: 9e5429f45cf9face9fed585d0233534993e36b58
 refs/heads/autosynth-bigtable-admin: 6379a2bc712f2736c83de0e009b4d26da4fa82ca
 refs/heads/autosynth-containeranalysis: af5b804492292b43372c9fe00386696136ccae89
-refs/heads/autosynth-datastore: aa5f27188cc55ebb5773b6597c6c49c965a40add
+refs/heads/autosynth-datastore: 77fb0ca8e507c38b4b559836b5c55576ef50e0c9
 refs/heads/autosynth-dialogflow: 7dbc2c1ea714328ccfa4f33645045f017ff080e7
 refs/heads/autosynth-errorreporting: 1101a04e8be074802c35332d5fcf8297f61cae32
 refs/heads/autosynth-firestore: d1a44f9acc302750e37b008ecb9c1aa535cc94df
diff --git a/branches/autosynth-datastore/google-cloud-clients/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/MessageDispatcher.java b/branches/autosynth-datastore/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() {
