---

sduskis · web-flow · commit d5cb520631d2 · 2019-04-18T10:31:56.000-04:00
yaml --- r: 27237 b: refs/heads/pubsub-ordering-keys c: 6be9e3d h: refs/heads/master i: 27235: 1b23581
diff --git a/[refs] b/[refs]
@@ -155,7 +155,7 @@ refs/tags/v0.72.0: a7703f2593ba312c0b2dde6fdfd4f5c764bb55ac
 refs/tags/v0.73.0: 21241ea8be9439cc5764c4944cdce21d34ce4f9e
 refs/tags/v0.74.0: 9d1f733dbbf790de7b494418523b69c4a9a57638
 refs/heads/ignoretest: 23c412ae07af3d0ab1caa2d44d5bc5c0ccb8b31d
-refs/heads/pubsub-ordering-keys: 7f96c1a0e5257639e61ac513ee06d7d6179cd8a7
+refs/heads/pubsub-ordering-keys: 6be9e3dd09d797dc60025f54fa376cb9625ce159
 "refs/heads/update_mvn_badge": ae2d773814db0f71197ccf5a8612ee1d8056f8de
 refs/tags/v0.75.0: c3673089ae09a897c1b4cf7dfe167fe4f8ab32fb
 refs/tags/v0.76.0: 395b016826d3ddf9cb8b34919636df15a4dbd032
diff --git a/branches/pubsub-ordering-keys/google-cloud-clients/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/SequentialExecutorService.java b/branches/pubsub-ordering-keys/google-cloud-clients/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/SequentialExecutorService.java
@@ -101,35 +101,29 @@ void execute(final String key, Runnable task) {
 
     protected abstract void execute(String key, Deque<Runnable> finalTasks);
 
-    /** Cancels every task in the queue assoicated with {@code key}. */
-    void cancelQueuedTasks(final String key, Throwable e) {
-      // TODO(kimkyung-goog): Ensure execute() fails once cancelQueueTasks() has been ever invoked,
-      // so that no more tasks are scheduled.
-      synchronized (tasksByKey) {
-        final Deque<Runnable> tasks = tasksByKey.get(key);
-        if (tasks == null) {
-          return;
-        }
-        while (!tasks.isEmpty()) {
-          Runnable task = tasks.poll();
-          if (task instanceof CancellableRunnable) {
-            ((CancellableRunnable) task).cancel(e);
-          } else {
-            logger.log(
-                Level.WARNING,
-                "Attempted to cancel Runnable that was not CancellableRunnable; ignored.");
-          }
-        }
-      }
-    }
-
     protected void invokeCallback(final Deque<Runnable> tasks) {
       // TODO(kimkyung-goog): Check if there is a race when task list becomes empty.
       Runnable task = tasks.poll();
       if (task != null) {
         task.run();
       }
     }
+  }
+
+  private static class AutoExecutor extends SequentialExecutor {
+    AutoExecutor(Executor executor) {
+      super(executor);
+    }
+
+    protected void execute(final String key, final Deque<Runnable> finalTasks) {
+      executor.execute(
+          new Runnable() {
+            @Override
+            public void run() {
+              invokeCallbackAndExecuteNext(key, finalTasks);
+            }
+          });
+    }
 
     protected void invokeCallbackAndExecuteNext(final String key, final Deque<Runnable> tasks) {
       invokeCallback(tasks);
@@ -153,22 +147,6 @@ public void run() {
     }
   }
 
-  private static class AutoExecutor extends SequentialExecutor {
-    AutoExecutor(Executor executor) {
-      super(executor);
-    }
-
-    protected void execute(final String key, final Deque<Runnable> finalTasks) {
-      executor.execute(
-          new Runnable() {
-            @Override
-            public void run() {
-              invokeCallbackAndExecuteNext(key, finalTasks);
-            }
-          });
-    }
-  }
-
   private static class CallbackExecutor extends SequentialExecutor {
     CallbackExecutor(Executor executor) {
       super(executor);
@@ -255,4 +233,26 @@ public void run() {
           });
     }
   }
+
+  /** Cancels every task in the queue assoicated with {@code key}. */
+  void cancelQueuedTasks(final String key, Throwable e) {
+    // TODO(kimkyung-goog): Ensure execute() fails once cancelQueueTasks() has been ever invoked,
+    // so that no more tasks are scheduled.
+    synchronized (tasksByKey) {
+      final Deque<Runnable> tasks = tasksByKey.get(key);
+      if (tasks == null) {
+        return;
+      }
+      while (!tasks.isEmpty()) {
+        Runnable task = tasks.poll();
+        if (task instanceof CancellableRunnable) {
+          ((CancellableRunnable) task).cancel(e);
+        } else {
+          logger.log(
+              Level.WARNING,
+              "Attempted to cancel Runnable that was not CancellableRunnable; ignored.");
+        }
+      }
+    }
+  }
 }
