---

sduskis · web-flow · commit 63defab6e4de · 2019-04-25T16:39:04.000-04:00
yaml --- r: 32745 b: refs/heads/autosynth-iot c: abf92a3 h: refs/heads/master i: 32743: 781d90e
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: 264643df7bcc715efbd648c3ab40389ae2253d19
+refs/heads/autosynth-iot: abf92a3ea30a7375790f6447532f6fd45bed5744
 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
@@ -40,13 +40,13 @@
 import com.google.cloud.pubsub.v1.stub.PublisherStub;
 import com.google.cloud.pubsub.v1.stub.PublisherStubSettings;
 import com.google.common.base.Preconditions;
-import com.google.common.collect.ImmutableList;
 import com.google.pubsub.v1.PublishRequest;
 import com.google.pubsub.v1.PublishResponse;
 import com.google.pubsub.v1.PubsubMessage;
 import com.google.pubsub.v1.TopicName;
 import com.google.pubsub.v1.TopicNames;
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Iterator;
 import java.util.LinkedList;
@@ -197,7 +197,7 @@ public ApiFuture<String> publish(PubsubMessage message) {
     }
 
     message = messageTransform.apply(message);
-    OutstandingBatch batchToSend = null;
+    List<OutstandingBatch> batchesToSend = new ArrayList<>();
     final OutstandingPublish outstandingPublish = new OutstandingPublish(message);
     messagesBatchLock.lock();
     try {
@@ -206,19 +206,18 @@ public ApiFuture<String> publish(PubsubMessage message) {
           && hasBatchingBytes()
           && messagesBatch.getBatchedBytes() + outstandingPublish.messageSize
               >= getMaxBatchBytes()) {
-        batchToSend = messagesBatch.popOutstandingBatch();
+        batchesToSend.add(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() || outstandingPublish.messageSize < getMaxBatchBytes()) {
-        messagesBatch.addMessage(outstandingPublish, outstandingPublish.messageSize);
+      messagesBatch.addMessage(outstandingPublish, outstandingPublish.messageSize);
 
-        // If after adding the message we have reached the batch max messages then we have a batch
-        // to send.
-        if (messagesBatch.getMessagesCount() == getBatchingSettings().getElementCountThreshold()) {
-          batchToSend = messagesBatch.popOutstandingBatch();
-        }
+      // 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());
       }
       // Setup the next duration based delivery alarm if there are messages batched.
       setupAlarm();
@@ -228,32 +227,17 @@ && hasBatchingBytes()
 
     messagesWaiter.incrementPendingMessages(1);
 
-    if (batchToSend != null) {
-      logger.log(Level.FINER, "Scheduling a batch for immediate sending.");
-      final OutstandingBatch finalBatchToSend = batchToSend;
-      executor.execute(
-          new Runnable() {
-            @Override
-            public void run() {
-              publishOutstandingBatch(finalBatchToSend);
-            }
-          });
-    }
-
-    // If the message is over the size limit, it was not added to the pending messages and it will
-    // be sent in its own batch immediately.
-    if (hasBatchingBytes() && outstandingPublish.messageSize >= getMaxBatchBytes()) {
-      logger.log(
-          Level.FINER, "Message exceeds the max batch bytes, scheduling it for immediate send.");
-      executor.execute(
-          new Runnable() {
-            @Override
-            public void run() {
-              publishOutstandingBatch(
-                  new OutstandingBatch(
-                      ImmutableList.of(outstandingPublish), outstandingPublish.messageSize));
-            }
-          });
+    if (!batchesToSend.isEmpty()) {
+      for (final OutstandingBatch batch : batchesToSend) {
+        logger.log(Level.FINER, "Scheduling a batch for immediate sending.");
+        executor.execute(
+            new Runnable() {
+              @Override
+              public void run() {
+                publishOutstandingBatch(batch);
+              }
+            });
+      }
     }
 
     return outstandingPublish.publishResult;
