---

sduskis · web-flow · commit e3548ae5e02b · 2019-04-16T11:19:35.000-04:00
yaml --- r: 31839 b: refs/heads/autosynth-datastore c: b1ebd2d h: refs/heads/master i: 31837: 08c42d7 31835: 83276c9 31831: 4ea6052 31823: 62a47bb 31807: 681f4b1
diff --git a/[refs] b/[refs]
@@ -127,7 +127,7 @@ refs/heads/autosynth-bigquerystorage: a345f72d7a6358e9fea9dcdae94ec85a27da3088
 refs/heads/autosynth-bigtable: 2a6af7e2959fd79c99b8ca5d773f29a6434b546d
 refs/heads/autosynth-bigtable-admin: 6379a2bc712f2736c83de0e009b4d26da4fa82ca
 refs/heads/autosynth-containeranalysis: 18d210f81f17cf74864d0db2c29e834302f74f2a
-refs/heads/autosynth-datastore: 79b64df4705b7a47895ec02aae4e84e5b8107884
+refs/heads/autosynth-datastore: b1ebd2d84384315f5788d1fa9f48c0a60265de20
 refs/heads/autosynth-dialogflow: 841930a680ebe370c9fc6ae824465f6bc51a5a46
 refs/heads/autosynth-errorreporting: 3f176c20b55dfaaa8fc32f28d82b31784b93e636
 refs/heads/autosynth-firestore: 983c75e4fb1076502c8cac73ef0538bdb10884f3
diff --git a/branches/autosynth-datastore/google-cloud-clients/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/Publisher.java b/branches/autosynth-datastore/google-cloud-clients/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/Publisher.java
@@ -86,8 +86,7 @@ public class Publisher {
   private final BatchingSettings batchingSettings;
 
   private final Lock messagesBatchLock;
-  private List<OutstandingPublish> messagesBatch;
-  private int batchedBytes;
+  private MessagesBatch messagesBatch;
 
   private final AtomicBoolean activeAlarm;
 
@@ -116,7 +115,7 @@ private Publisher(Builder builder) throws IOException {
     this.batchingSettings = builder.batchingSettings;
     this.messageTransform = builder.messageTransform;
 
-    messagesBatch = new LinkedList<>();
+    messagesBatch = new MessagesBatch();
     messagesBatchLock = new ReentrantLock();
     activeAlarm = new AtomicBoolean(false);
     executor = builder.executorProvider.getExecutor();
@@ -207,24 +206,19 @@ public ApiFuture<String> publish(PubsubMessage message) {
       // Check if the next message makes the current batch exceed the max batch byte size.
       if (!messagesBatch.isEmpty()
           && hasBatchingBytes()
-          && batchedBytes + messageSize >= getMaxBatchBytes()) {
-        batchToSend = new OutstandingBatch(messagesBatch, batchedBytes);
-        messagesBatch = new LinkedList<>();
-        batchedBytes = 0;
+          && messagesBatch.getBatchedBytes() + messageSize >= getMaxBatchBytes()) {
+        batchToSend = messagesBatch.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()) {
-        batchedBytes += messageSize;
-        messagesBatch.add(outstandingPublish);
+        messagesBatch.addMessage(outstandingPublish, messageSize);
 
         // If after adding the message we have reached the batch max messages then we have a batch
         // to send.
-        if (messagesBatch.size() == getBatchingSettings().getElementCountThreshold()) {
-          batchToSend = new OutstandingBatch(messagesBatch, batchedBytes);
-          messagesBatch = new LinkedList<>();
-          batchedBytes = 0;
+        if (messagesBatch.getMessagesCount() == getBatchingSettings().getElementCountThreshold()) {
+          batchToSend = messagesBatch.popOutstandingBatch();
         }
       }
       // Setup the next duration based delivery alarm if there are messages batched.
@@ -303,9 +297,7 @@ public void publishAllOutstanding() {
       if (messagesBatch.isEmpty()) {
         return;
       }
-      batchToSend = new OutstandingBatch(messagesBatch, batchedBytes);
-      messagesBatch = new LinkedList<>();
-      batchedBytes = 0;
+      batchToSend = messagesBatch.popOutstandingBatch();
     } finally {
       messagesBatchLock.unlock();
     }
@@ -640,4 +632,37 @@ public Publisher build() throws IOException {
       return new Publisher(this);
     }
   }
+
+  private static class MessagesBatch {
+    private List<OutstandingPublish> messages = new LinkedList<>();
+    private int batchedBytes;
+
+    private OutstandingBatch popOutstandingBatch() {
+      OutstandingBatch batch = new OutstandingBatch(messages, batchedBytes);
+      reset();
+      return batch;
+    }
+
+    private void reset() {
+      messages = new LinkedList<>();
+      batchedBytes = 0;
+    }
+
+    private boolean isEmpty() {
+      return messages.isEmpty();
+    }
+
+    private int getBatchedBytes() {
+      return batchedBytes;
+    }
+
+    private void addMessage(OutstandingPublish message, int messageSize) {
+      messages.add(message);
+      batchedBytes += messageSize;
+    }
+
+    private int getMessagesCount() {
+      return messages.size();
+    }
+  }
 }
