chore: move childRefs to AccumulatingRead instead of BlobDescriptorStreamRead

Author: BenWhitehead

google-cloud-storage/src/main/java/com/google/cloud/storage/BlobDescriptorStreamRead.java

@@ -35,32 +35,23 @@ abstract class BlobDescriptorStreamRead implements AutoCloseable, Closeable {
 
   protected final long readId;
   protected final RangeSpec rangeSpec;
-  protected final List<ChildRef> childRefs;
   protected final RetryContext retryContext;
   protected final AtomicLong readOffset;
   protected boolean closed;
   protected boolean tombstoned;
 
   BlobDescriptorStreamRead(long readId, RangeSpec rangeSpec, RetryContext retryContext) {
-    this(
-        readId,
-        rangeSpec,
-        Collections.synchronizedList(new ArrayList<>()),
-        new AtomicLong(rangeSpec.begin()),
-        retryContext,
-        false);
+    this(readId, rangeSpec, new AtomicLong(rangeSpec.begin()), retryContext, false);
   }
 
   BlobDescriptorStreamRead(
       long readId,
       RangeSpec rangeSpec,
-      List<ChildRef> childRefs,
       AtomicLong readOffset,
       RetryContext retryContext,
       boolean closed) {
     this.readId = readId;
     this.rangeSpec = rangeSpec;
-    this.childRefs = childRefs;
     this.retryContext = retryContext;
     this.readOffset = readOffset;
     this.closed = closed;
@@ -99,17 +90,15 @@ final ReadRange makeReadRange() {
   }
 
   @Override
-  public void close() throws IOException {
-    if (!closed) {
-      retryContext.reset();
-      closed = true;
-      GrpcUtils.closeAll(childRefs);
-    }
-  }
+  public abstract void close() throws IOException;
 
-  abstract <T extends Throwable> void recordError(T t, OnSuccess onSuccess, OnFailure<T> onFailure);
+  <T extends Throwable> void recordError(T t, OnSuccess onSuccess, OnFailure<T> onFailure) {
+    retryContext.recordError(t, onSuccess, onFailure);
+  }
 
-  public abstract boolean readyToSend();
+  boolean readyToSend() {
+    return !tombstoned && !retryContext.inBackoff();
+  }
 
   static AccumulatingRead<byte[]> createByteArrayAccumulatingRead(
       long readId,
@@ -129,6 +118,7 @@ static ZeroCopyByteStringAccumulatingRead createZeroCopyByteStringAccumulatingRe
 
   /** Base class of a read that will accumulate before completing by resolving a future */
   abstract static class AccumulatingRead<Result> extends BlobDescriptorStreamRead {
+    protected final List<ChildRef> childRefs;
     protected final SettableApiFuture<Result> complete;
 
     private AccumulatingRead(
@@ -138,6 +128,7 @@ private AccumulatingRead(
         SettableApiFuture<Result> complete) {
       super(readId, rangeSpec, retryContext);
       this.complete = complete;
+      this.childRefs = Collections.synchronizedList(new ArrayList<>());
     }
 
     private AccumulatingRead(
@@ -148,7 +139,8 @@ private AccumulatingRead(
         RetryContext retryContext,
         boolean closed,
         SettableApiFuture<Result> complete) {
-      super(readId, rangeSpec, childRefs, readOffset, retryContext, closed);
+      super(readId, rangeSpec, readOffset, retryContext, closed);
+      this.childRefs = childRefs;
       this.complete = complete;
     }
 
@@ -157,6 +149,14 @@ boolean acceptingBytes() {
       return !complete.isDone() && !tombstoned;
     }
 
+    @Override
+    void accept(ChildRef childRef) throws IOException {
+      retryContext.reset();
+      int size = childRef.byteString().size();
+      childRefs.add(childRef);
+      readOffset.addAndGet(size);
+    }
+
     @Override
     ApiFuture<?> fail(Throwable t) {
       try {
@@ -171,13 +171,12 @@ ApiFuture<?> fail(Throwable t) {
     }
 
     @Override
-    <T extends Throwable> void recordError(T t, OnSuccess onSuccess, OnFailure<T> onFailure) {
-      retryContext.recordError(t, onSuccess, onFailure);
-    }
-
-    @Override
-    public boolean readyToSend() {
-      return !tombstoned && !retryContext.inBackoff();
+    public void close() throws IOException {
+      if (!closed) {
+        retryContext.reset();
+        closed = true;
+        GrpcUtils.closeAll(childRefs);
+      }
     }
   }
 
@@ -193,11 +192,10 @@ private StreamingRead(long readId, RangeSpec range, RetryContext retryContext) {
     private StreamingRead(
         long readId,
         RangeSpec rangeSpec,
-        List<ChildRef> childRefs,
         AtomicLong readOffset,
         RetryContext retryContext,
         boolean closed) {
-      super(readId, rangeSpec, childRefs, readOffset, retryContext, closed);
+      super(readId, rangeSpec, readOffset, retryContext, closed);
     }
   }
 
@@ -222,14 +220,6 @@ private ByteArrayAccumulatingRead(
       super(readId, rangeSpec, childRefs, readOffset, retryContext, closed, complete);
     }
 
-    @Override
-    void accept(ChildRef childRef) throws IOException {
-      retryContext.reset();
-      int size = childRef.byteString().size();
-      childRefs.add(childRef);
-      readOffset.addAndGet(size);
-    }
-
     @Override
     void eof() throws IOException {
       retryContext.reset();
@@ -283,14 +273,6 @@ public ByteString byteString() {
       return byteString;
     }
 
-    @Override
-    void accept(ChildRef childRef) throws IOException {
-      retryContext.reset();
-      int size = childRef.byteString().size();
-      childRefs.add(childRef);
-      readOffset.addAndGet(size);
-    }
-
     @Override
     void eof() throws IOException {
       retryContext.reset();

google-cloud-storage/src/test/java/com/google/cloud/storage/BlobDescriptorStreamTest.java

@@ -44,9 +44,8 @@
 import com.google.storage.v2.BidiReadObjectSpec;
 import com.google.storage.v2.BucketName;
 import com.google.storage.v2.Object;
+import java.io.IOException;
 import java.nio.channels.AsynchronousCloseException;
-import java.util.ArrayList;
-import java.util.List;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
@@ -132,8 +131,7 @@ public static void afterClass() throws Exception {
   public void streamRestartShouldNotSendARequestIfAllReadsAreInBackoff() {
     RetryContext read1RetryContext = retryContextProvider.create();
     TestBlobDescriptorStreamRead read1 =
-        new TestBlobDescriptorStreamRead(
-            1, RangeSpec.of(1, 2), new ArrayList<>(), read1RetryContext);
+        new TestBlobDescriptorStreamRead(1, RangeSpec.of(1, 2), read1RetryContext);
     state.putOutstandingRead(1, read1);
 
     RetryContext streamRetryContext = retryContextProvider.create();
@@ -152,8 +150,7 @@ public void streamRestartShouldNotSendARequestIfAllReadsAreInBackoff() {
   public void streamRestartShouldSendARequestIfReadsAreNotInBackoff() {
     RetryContext read1RetryContext = retryContextProvider.create();
     TestBlobDescriptorStreamRead read1 =
-        new TestBlobDescriptorStreamRead(
-            1, RangeSpec.of(1, 2), new ArrayList<>(), read1RetryContext);
+        new TestBlobDescriptorStreamRead(1, RangeSpec.of(1, 2), read1RetryContext);
     read1.readyToSend = true;
     state.putOutstandingRead(1, read1);
 
@@ -238,9 +235,8 @@ private static class TestBlobDescriptorStreamRead extends BlobDescriptorStreamRe
     private boolean readyToSend = false;
     private final SettableApiFuture<Throwable> fail = SettableApiFuture.create();
 
-    TestBlobDescriptorStreamRead(
-        long readId, RangeSpec rangeSpec, List<ChildRef> childRefs, RetryContext retryContext) {
-      super(readId, rangeSpec, childRefs, new AtomicLong(rangeSpec.begin()), retryContext, false);
+    TestBlobDescriptorStreamRead(long readId, RangeSpec rangeSpec, RetryContext retryContext) {
+      super(readId, rangeSpec, new AtomicLong(rangeSpec.begin()), retryContext, false);
     }
 
     @Override
@@ -273,10 +269,12 @@ public boolean readyToSend() {
       return readyToSend;
     }
 
+    @Override
+    public void close() throws IOException {}
+
     static TestBlobDescriptorStreamRead of() {
       long id = readIdSeq.getAndIncrement();
-      return new TestBlobDescriptorStreamRead(
-          id, RangeSpec.of(0, 10), new ArrayList<>(), RetryContext.neverRetry());
+      return new TestBlobDescriptorStreamRead(id, RangeSpec.of(0, 10), RetryContext.neverRetry());
     }
   }
 }