Refactoring of the Pub/Sub Ordering keys client (#4962)

googleapis · Apr 17, 2019 · 7732506 · 7732506
1 parent e88432a
commit 7732506
Show file tree

Hide file tree

Showing 2 changed files with 19 additions and 17 deletions.
diff --git a/...cloud-clients/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/Publisher.java b/...cloud-clients/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/Publisher.java
@@ -234,30 +234,30 @@ public ApiFuture<String> publish(PubsubMessage message) {
     messagesBatchLock.lock();
     try {
       // Check if the next message makes the current batch exceed the max batch byte size.
-      MessagesBatch batch = messagesBatches.get(orderingKey);
-      if (batch == null) {
-        batch = new MessagesBatch(orderingKey);
-        messagesBatches.put(orderingKey, batch);
+      MessagesBatch messageBatch = messagesBatches.get(orderingKey);
+      if (messageBatch == null) {
+        messageBatch = new MessagesBatch(orderingKey);
+        messagesBatches.put(orderingKey, messageBatch);
       }
-      if (!batch.isEmpty()
+      if (!messageBatch.isEmpty()
           && hasBatchingBytes()
-          && batch.getBatchedBytes() + messageSize >= getMaxBatchBytes()) {
-        batchToSend = batch.popOutstandingBatch();
+          && messageBatch.getBatchedBytes() + messageSize >= getMaxBatchBytes()) {
+        batchToSend = messageBatch.popOutstandingBatch();
       }
 
       // Border case if the message to send is greater or equals to the max batch size then can't
       // be included in the current batch and instead sent immediately.
       if (!hasBatchingBytes() || messageSize < getMaxBatchBytes()) {
-        batch.addMessage(outstandingPublish, messageSize);
+        messageBatch.addMessage(outstandingPublish, messageSize);
         // If after adding the message we have reached the batch max messages then we have a batch
         // to send.
-        if (batch.getMessagesCount() == getBatchingSettings().getElementCountThreshold()) {
-          batchToSend = batch.popOutstandingBatch();
+        if (messageBatch.getMessagesCount() == getBatchingSettings().getElementCountThreshold()) {
+          batchToSend = messageBatch.popOutstandingBatch();
         }
       }
 
       // Setup the next duration based delivery alarm if there are messages batched.
-      if (!batch.isEmpty()) {
+      if (!messageBatch.isEmpty()) {
         setupDurationBasedPublishAlarm();
       } else {
         messagesBatches.remove(orderingKey);
@@ -405,7 +405,9 @@ public ApiFuture call() {
             }
           };
       ApiFutures.addCallback(
-          sequentialExecutor.submit(outstandingBatch.orderingKey, func), futureCallback);
+          sequentialExecutor.submit(outstandingBatch.orderingKey, func),
+          futureCallback,
+          directExecutor());
     }
   }
 

diff --git a/...ogle-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/SequentialExecutorService.java b/...ogle-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/SequentialExecutorService.java
@@ -102,7 +102,7 @@ public void cancel(Throwable e) {
   }
 
   /** Runs synchronous {@code Runnable} tasks sequentially. */
-  public void submit(final String key, final Runnable runnable) {
+  void submit(final String key, final Runnable runnable) {
     autoSequentialExecutor.execute(key, runnable);
   }
 
@@ -128,7 +128,7 @@ enum TaskCompleteAction {
      * Creates a AutoSequentialExecutor which executes the next queued task automatically when the
      * previous task has completed.
      */
-    public static SequentialExecutor newAutoSequentialExecutor(Executor executor) {
+    static SequentialExecutor newAutoSequentialExecutor(Executor executor) {
       return new SequentialExecutor(executor, TaskCompleteAction.EXECUTE_NEXT_TASK);
     }
 
@@ -147,7 +147,7 @@ private SequentialExecutor(Executor executor, TaskCompleteAction taskCompleteAct
       this.tasksByKey = new HashMap<>();
     }
 
-    public void execute(final String key, Runnable task) {
+    void execute(final String key, Runnable task) {
       Deque<Runnable> newTasks;
       synchronized (tasksByKey) {
         newTasks = tasksByKey.get(key);
@@ -182,7 +182,7 @@ public void run() {
     }
 
     /** Cancels every task in the queue assoicated with {@code key}. */
-    public void cancelQueuedTasks(final String key, Throwable e) {
+    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) {
@@ -204,7 +204,7 @@ public void cancelQueuedTasks(final String key, Throwable e) {
     }
 
     /** Executes the next queued task associated with {@code key}. */
-    public void resume(final String key) {
+    void resume(final String key) {
       if (taskCompleteAction.equals(TaskCompleteAction.EXECUTE_NEXT_TASK)) {
         // resume() is no-op since tasks are executed automatically.
         return;