---

sduskis · web-flow · commit 7cd9967ded49 · 2019-04-30T11:44:16.000-04:00
yaml --- r: 32755 b: refs/heads/autosynth-iot c: 61b9e4f h: refs/heads/master i: 32753: 340af83 32751: 740f58e
diff --git a/[refs] b/[refs]
@@ -131,7 +131,7 @@ refs/heads/autosynth-datastore: f1efc3dc465174f41041acd56cf29badcec3e5bd
 refs/heads/autosynth-dialogflow: 73974cc32e5212aec0126472e0bc1442886fedaf
 refs/heads/autosynth-errorreporting: effe8001d110ad584187b30aafc473db0dd4a15f
 refs/heads/autosynth-firestore: e79eeb26930dfae4439424ad2fda5874eeca54c8
-refs/heads/autosynth-iot: 59c609fc9dd9d865a1466969410e280b7bdf62dd
+refs/heads/autosynth-iot: 61b9e4fc725fda428fa6ec8ffe177cd86bf230a9
 refs/heads/autosynth-kms: 6b65b0f34c12d141031c7288cdc01e550212d0f6
 refs/heads/autosynth-language: e73905aa7672afa47240e65b25c087207f4594f9
 refs/heads/autosynth-os-login: 123ba209c5769d0ee067e0ce5848bec13b42a4f4
diff --git a/branches/autosynth-iot/google-cloud-clients/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/Publisher.java b/branches/autosynth-iot/google-cloud-clients/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/Publisher.java
@@ -116,7 +116,7 @@ private Publisher(Builder builder) throws IOException {
     this.batchingSettings = builder.batchingSettings;
     this.messageTransform = builder.messageTransform;
 
-    messagesBatch = new MessagesBatch();
+    messagesBatch = new MessagesBatch(batchingSettings);
     messagesBatchLock = new ReentrantLock();
     activeAlarm = new AtomicBoolean(false);
     executor = builder.executorProvider.getExecutor();
@@ -196,29 +196,12 @@ public ApiFuture<String> publish(PubsubMessage message) {
       throw new IllegalStateException("Cannot publish on a shut-down publisher.");
     }
 
-    message = messageTransform.apply(message);
-    List<OutstandingBatch> batchesToSend = new ArrayList<>();
-    final OutstandingPublish outstandingPublish = new OutstandingPublish(message);
+    final OutstandingPublish outstandingPublish =
+        new OutstandingPublish(messageTransform.apply(message));
+    List<OutstandingBatch> batchesToSend;
     messagesBatchLock.lock();
     try {
-      // Check if the next message makes the current batch exceed the max batch byte size.
-      if (!messagesBatch.isEmpty()
-          && hasBatchingBytes()
-          && messagesBatch.getBatchedBytes() + outstandingPublish.messageSize
-              >= getMaxBatchBytes()) {
-        batchesToSend.add(messagesBatch.popOutstandingBatch());
-      }
-
-      messagesBatch.addMessage(outstandingPublish, outstandingPublish.messageSize);
-
-      // Border case: If the message to send is greater or equals to the max batch size then send it
-      // immediately.
-      // Alternatively if after adding the message we have reached the batch max messages then we
-      // have a batch to send.
-      if ((hasBatchingBytes() && outstandingPublish.messageSize >= getMaxBatchBytes())
-          || messagesBatch.getMessagesCount() == getBatchingSettings().getElementCountThreshold()) {
-        batchesToSend.add(messagesBatch.popOutstandingBatch());
-      }
+      batchesToSend = messagesBatch.add(outstandingPublish);
       // Setup the next duration based delivery alarm if there are messages batched.
       setupAlarm();
     } finally {
@@ -378,10 +361,6 @@ public BatchingSettings getBatchingSettings() {
     return batchingSettings;
   }
 
-  private long getMaxBatchBytes() {
-    return getBatchingSettings().getRequestByteThreshold();
-  }
-
   /**
    * Schedules immediate publishing of any outstanding messages and waits until all are processed.
    *
@@ -410,10 +389,6 @@ public boolean awaitTermination(long duration, TimeUnit unit) throws Interrupted
     return backgroundResources.awaitTermination(duration, unit);
   }
 
-  private boolean hasBatchingBytes() {
-    return getMaxBatchBytes() > 0;
-  }
-
   /**
    * Constructs a new {@link Builder} using the given topic.
    *
@@ -616,8 +591,14 @@ public Publisher build() throws IOException {
   }
 
   private static class MessagesBatch {
-    private List<OutstandingPublish> messages = new LinkedList<>();
+    private List<OutstandingPublish> messages;
     private int batchedBytes;
+    private final BatchingSettings batchingSettings;
+
+    public MessagesBatch(BatchingSettings batchingSettings) {
+      this.batchingSettings = batchingSettings;
+      reset();
+    }
 
     private OutstandingBatch popOutstandingBatch() {
       OutstandingBatch batch = new OutstandingBatch(messages, batchedBytes);
@@ -638,13 +619,40 @@ private int getBatchedBytes() {
       return batchedBytes;
     }
 
-    private void addMessage(OutstandingPublish message, int messageSize) {
-      messages.add(message);
-      batchedBytes += messageSize;
-    }
-
     private int getMessagesCount() {
       return messages.size();
     }
+
+    private boolean hasBatchingBytes() {
+      return getMaxBatchBytes() > 0;
+    }
+
+    private long getMaxBatchBytes() {
+      return batchingSettings.getRequestByteThreshold();
+    }
+
+    private List<OutstandingBatch> add(OutstandingPublish outstandingPublish) {
+      List<OutstandingBatch> batchesToSend = new ArrayList<>();
+      // Check if the next message makes the current batch exceed the max batch byte size.
+      if (!isEmpty()
+          && hasBatchingBytes()
+          && getBatchedBytes() + outstandingPublish.messageSize >= getMaxBatchBytes()) {
+        batchesToSend.add(popOutstandingBatch());
+      }
+
+      messages.add(outstandingPublish);
+      batchedBytes += outstandingPublish.messageSize;
+
+      // Border case: If the message to send is greater or equals to the max batch size then send it
+      // immediately.
+      // Alternatively if after adding the message we have reached the batch max messages then we
+      // have a batch to send.
+      if ((hasBatchingBytes() && outstandingPublish.messageSize >= getMaxBatchBytes())
+          || getMessagesCount() == batchingSettings.getElementCountThreshold()) {
+        batchesToSend.add(popOutstandingBatch());
+      }
+
+      return batchesToSend;
+    }
   }
 }
