Updates from feedback

garrettjonesgoogle · Mar 25, 2016 · 7d7e315 · 7d7e315
1 parent 38d978f
commit 7d7e315
Show file tree

Hide file tree

Showing 3 changed files with 17 additions and 14 deletions.
diff --git a/src/main/java/com/google/api/gax/bundling/AccumulatingBundleReceiver.java b/src/main/java/com/google/api/gax/bundling/AccumulatingBundleReceiver.java
@@ -48,9 +48,7 @@ public void validateItem(T message) {
 
   @Override
   public void processBundle(List<T> bundle) {
-    List<T> bundleCopy = new ArrayList<>();
-    bundleCopy.addAll(bundle);
-    bundles.add(bundleCopy);
+    bundles.add(new ArrayList<>(bundle));
   }
 
   /**

diff --git a/src/main/java/com/google/api/gax/bundling/NumericThreshold.java b/src/main/java/com/google/api/gax/bundling/NumericThreshold.java
@@ -49,7 +49,8 @@ public class NumericThreshold<E> implements BundlingThreshold<E> {
    * Constructs a NumericThreshold.
    *
    * @param threshold The value that allows an event to happen.
-   * @param limit The value that forces an event to happen.
+   * @param limit The value that forces an event to happen. If null, then this is
+   * not enforced.
    * @param extractor Object that extracts a numeric value from the value object.
    */
   public NumericThreshold(long threshold, @Nullable Long limit, ElementCounter<E> extractor) {

diff --git a/src/main/java/com/google/api/gax/bundling/ThresholdBundler.java b/src/main/java/com/google/api/gax/bundling/ThresholdBundler.java
@@ -151,38 +151,34 @@ public ThresholdBundleHandle add(E e) {
     final Lock lock = this.lock;
     lock.lock();
     try {
-      for (BundlingThreshold<E> threshold : thresholdPrototypes) {
-        if (!threshold.canAccept(e)) {
-          throw new IllegalArgumentException("Single item too large for bundle");
-        }
-      }
+      validateLimits(e);
 
-      boolean signal = false;
+      boolean signalBundleIsReady = false;
       Bundle bundleOfAddedItem = null;
       if (currentOpenBundle == null) {
         currentOpenBundle = new Bundle(thresholdPrototypes, externalThresholdPrototypes, maxDelay);
         currentOpenBundle.start();
-        signal = true;
+        signalBundleIsReady = true;
       }
 
       if (currentOpenBundle.canAccept(e)) {
         currentOpenBundle.add(e);
         bundleOfAddedItem = currentOpenBundle;
         if (currentOpenBundle.isAnyThresholdReached()) {
-          signal = true;
+          signalBundleIsReady = true;
           closedBundles.add(currentOpenBundle);
           currentOpenBundle = null;
         }
       } else {
-        signal = true;
+        signalBundleIsReady = true;
         closedBundles.add(currentOpenBundle);
         currentOpenBundle = new Bundle(thresholdPrototypes, externalThresholdPrototypes, maxDelay);
         currentOpenBundle.start();
         currentOpenBundle.add(e);
         bundleOfAddedItem = currentOpenBundle;
       }
 
-      if (signal) {
+      if (signalBundleIsReady) {
         bundleCondition.signalAll();
       }
       return bundleOfAddedItem;
@@ -282,6 +278,14 @@ public boolean isEmpty() {
     }
   }
 
+  private void validateLimits(E e) {
+    for (BundlingThreshold<E> threshold : thresholdPrototypes) {
+      if (!threshold.canAccept(e)) {
+        throw new IllegalArgumentException("Single item too large for bundle");
+      }
+    }
+  }
+
   private boolean shouldWait() {
     if (closedBundles.size() > 0) {
       return false;