---

sduskis · sduskis · commit 516ec9b8e2f4 · 2019-04-24T11:58:18.000-04:00
yaml --- r: 27259 b: refs/heads/pubsub-ordering-keys c: 75acad9 h: refs/heads/master i: 27257: ad1da67 27255: 223267a
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: 956537c5ae22171b198568b729a32e42059db9ae
+refs/heads/pubsub-ordering-keys: 75acad92e69422d9ed62b6ea738905538ffcbd32
 "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/Publisher.java b/branches/pubsub-ordering-keys/google-cloud-clients/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/Publisher.java
@@ -390,13 +390,15 @@ public void onFailure(Throwable t) {
       ApiFutures.addCallback(publishCall(outstandingBatch), futureCallback, directExecutor());
     } else {
       // If ordering key is specified, publish the batch using the sequential executor.
-      sequentialExecutor.submit(outstandingBatch.orderingKey, new Callable<ApiFuture<PublishResponse>>() {
-        public ApiFuture<PublishResponse> call() {
-          ApiFuture<PublishResponse> future = publishCall(outstandingBatch);
-          ApiFutures.addCallback(future, futureCallback, directExecutor());
-          return future;
-        }
-      });
+      sequentialExecutor.submit(
+          outstandingBatch.orderingKey,
+          new Callable<ApiFuture<PublishResponse>>() {
+            public ApiFuture<PublishResponse> call() {
+              ApiFuture<PublishResponse> future = publishCall(outstandingBatch);
+              ApiFutures.addCallback(future, futureCallback, directExecutor());
+              return future;
+            }
+          });
     }
   }
 
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
@@ -29,7 +29,6 @@
 import java.util.Queue;
 import java.util.concurrent.Callable;
 import java.util.concurrent.CancellationException;
-import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.Executor;
 
 interface CancellableRunnable extends Runnable {
@@ -89,20 +88,22 @@ protected void callNextTaskAsync(final String key) {
         }
       }
       if (executeTask) {
-        executor.execute(new Runnable() {
-          @Override public void run() {
-            R task = null;
-            synchronized (tasksByKey) {
-              Queue<R> tasks = tasksByKey.get(key);
-              if (tasks != null && !tasks.isEmpty()) {
-                task = tasks.poll();
+        executor.execute(
+            new Runnable() {
+              @Override
+              public void run() {
+                R task = null;
+                synchronized (tasksByKey) {
+                  Queue<R> tasks = tasksByKey.get(key);
+                  if (tasks != null && !tasks.isEmpty()) {
+                    task = tasks.poll();
+                  }
+                }
+                if (task != null) {
+                  task.run();
+                }
               }
-            }
-            if (task != null) {
-              task.run();
-            }
-          }
-        });
+            });
       }
     }
   }
@@ -115,15 +116,18 @@ static class AutoExecutor extends SequentialExecutor<Runnable> {
 
     /** Runs synchronous {@code Runnable} tasks sequentially. */
     void submit(final String key, final Runnable task) {
-      super.execute(key, new Runnable() {
-        @Override public void run() {
-          try {
-            task.run();
-          } finally {
-            callNextTaskAsync(key);
-          }
-        }
-      });
+      super.execute(
+          key,
+          new Runnable() {
+            @Override
+            public void run() {
+              try {
+                task.run();
+              } finally {
+                callNextTaskAsync(key);
+              }
+            }
+          });
     }
   }
 
@@ -133,8 +137,9 @@ void submit(final String key, final Runnable task) {
    */
   @BetaApi
   static class CallbackExecutor extends SequentialExecutor<CancellableRunnable> {
-    static CancellationException CANCELLATION_EXCEPTION = new CancellationException(
-        "Execution cancelled because executing previous runnable failed.");
+    static CancellationException CANCELLATION_EXCEPTION =
+        new CancellationException(
+            "Execution cancelled because executing previous runnable failed.");
 
     CallbackExecutor(Executor executor) {
       super(executor);
@@ -177,41 +182,47 @@ <T> ApiFuture<T> submit(final String key, final Callable<ApiFuture<T>> callable)
 
       // Step 2: create the CancellableRunnable
       // Step 3: add the task to queue via `execute`
-      CancellableRunnable task = new CancellableRunnable() {
-        private boolean cancelled = false;
-
-        @Override public void run() {
-          // the task was cancelled
-          if (cancelled) {
-            return;
-          }
-
-          try {
-            // Step 4: call the `Callable`
-            ApiFutureCallback<T> callback = new ApiFutureCallback<T>() {
-              // Step 5.1: on success
-              @Override public void onSuccess(T msg) {
-                future.set(msg);
-                callNextTaskAsync(key);
+      CancellableRunnable task =
+          new CancellableRunnable() {
+            private boolean cancelled = false;
+
+            @Override
+            public void run() {
+              // the task was cancelled
+              if (cancelled) {
+                return;
               }
 
-              // Step 5.2: on failure
-              @Override public void onFailure(Throwable e) {
-                future.setException(e);
-                cancelQueuedTasks(key, CANCELLATION_EXCEPTION);
+              try {
+                // Step 4: call the `Callable`
+                ApiFutureCallback<T> callback =
+                    new ApiFutureCallback<T>() {
+                      // Step 5.1: on success
+                      @Override
+                      public void onSuccess(T msg) {
+                        future.set(msg);
+                        callNextTaskAsync(key);
+                      }
+
+                      // Step 5.2: on failure
+                      @Override
+                      public void onFailure(Throwable e) {
+                        future.setException(e);
+                        cancelQueuedTasks(key, CANCELLATION_EXCEPTION);
+                      }
+                    };
+                ApiFutures.addCallback(callable.call(), callback, directExecutor());
+              } catch (Exception e) {
+                cancel(e);
               }
-            };
-            ApiFutures.addCallback(callable.call(), callback, directExecutor());
-          } catch (Exception e) {
-            cancel(e);
-          }
-        }
+            }
 
-        @Override public void cancel(Throwable e) {
-          this.cancelled = true;
-          future.setException(e);
-        }
-      };
+            @Override
+            public void cancel(Throwable e) {
+              this.cancelled = true;
+              future.setException(e);
+            }
+          };
       execute(key, task);
       return future;
     }
