---

sduskis · web-flow · commit 58af4bcf0a3e · 2019-04-30T11:44:16.000-04:00
yaml --- r: 33659 b: refs/heads/autosynth-redis c: 61b9e4f h: refs/heads/master i: 33657: 3dbf74a 33655: 1e22a88
diff --git a/[refs] b/[refs]
@@ -135,7 +135,7 @@ refs/heads/autosynth-iot: 044be280805a59e06d09658688c9ee474a9815ad
 refs/heads/autosynth-kms: d31449d6621a50fb16a4bef4f30f0f3051d27d7c
 refs/heads/autosynth-language: 6130869312f99a1e7d3aa0485759172a23333cc5
 refs/heads/autosynth-os-login: 49028d40ac477fca5f948cc5a3ce7422729fdb67
-refs/heads/autosynth-redis: 59c609fc9dd9d865a1466969410e280b7bdf62dd
+refs/heads/autosynth-redis: 61b9e4fc725fda428fa6ec8ffe177cd86bf230a9
 refs/heads/autosynth-scheduler: 57f9fdb1e7de30c85f4ec7198931a07f50603e55
 refs/heads/autosynth-spanner: de02ca32edea133b68b51052e325359a3704b5d2
 refs/heads/autosynth-speech: 64692f6db11364f663921be02c08072b966b6e7b
diff --git a/branches/autosynth-redis/google-cloud-clients/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/Publisher.java b/branches/autosynth-redis/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;
+    }
   }
 }
