chore: gut BlobReadChannel

In order to facilitate migrating any RestorableState<ReadChannel> customers might have, we leave the existing class hierarchy in place and update BlobReadChannel.StateImpl#restore() to produce a new BlobReadChannelV2.

In the next major version this compatibility path will be removed.

To test compatibility a serialized instance of a BlobReadChannel from v2.16.0 has been created and serialized into blobWriteChannel.ser.properties along with a comment describing how the ser was generated.

Author: BenWhitehead

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

@@ -16,262 +16,105 @@
 
 package com.google.cloud.storage;
 
-import static com.google.cloud.RetryHelper.runWithRetries;
-
-import com.google.api.client.util.Preconditions;
-import com.google.api.gax.retrying.ResultRetryAlgorithm;
 import com.google.api.services.storage.model.StorageObject;
 import com.google.cloud.ReadChannel;
 import com.google.cloud.RestorableState;
-import com.google.cloud.RetryHelper;
-import com.google.cloud.Tuple;
+import com.google.cloud.storage.BlobReadChannelV2.BlobReadChannelContext;
 import com.google.cloud.storage.spi.v1.StorageRpc;
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.MoreObjects;
 import java.io.IOException;
 import java.io.Serializable;
 import java.nio.ByteBuffer;
-import java.nio.channels.ClosedChannelException;
 import java.util.Map;
 import java.util.Objects;
 
-/** Default implementation for ReadChannel. */
+/**
+ * Hierarchy retained for {@link RestorableState#restore()}. Will be removed in next major version!
+ */
+@Deprecated
 class BlobReadChannel implements ReadChannel {
 
-  private static final int DEFAULT_CHUNK_SIZE = 2 * 1024 * 1024;
-
-  private final HttpStorageOptions serviceOptions;
-  private final BlobId blob;
-  private final Map<StorageRpc.Option, ?> requestOptions;
-  private final HttpRetryAlgorithmManager retryAlgorithmManager;
-  private String lastEtag;
-  private long position;
-  private boolean isOpen;
-  private boolean endOfStream;
-  private int chunkSize = DEFAULT_CHUNK_SIZE;
-
-  private final StorageRpc storageRpc;
-  private final StorageObject storageObject;
-  private int bufferPos;
-  private byte[] buffer;
-  private long limit;
-
-  BlobReadChannel(
-      HttpStorageOptions serviceOptions, BlobId blob, Map<StorageRpc.Option, ?> requestOptions) {
-    this.serviceOptions = serviceOptions;
-    this.blob = blob;
-    this.requestOptions = requestOptions;
-    this.retryAlgorithmManager = serviceOptions.getRetryAlgorithmManager();
-    isOpen = true;
-    storageRpc = serviceOptions.getStorageRpcV1();
-    storageObject = Conversions.apiary().blobId().encode(blob);
-    this.limit = Long.MAX_VALUE;
-  }
+  private BlobReadChannel() {}
 
   @Override
   public RestorableState<ReadChannel> capture() {
-    StateImpl.Builder builder =
-        StateImpl.builder(serviceOptions, blob, requestOptions)
-            .setPosition(position)
-            .setIsOpen(isOpen)
-            .setEndOfStream(endOfStream)
-            .setChunkSize(chunkSize)
-            .setLimit(limit);
-    if (buffer != null) {
-      builder.setPosition(position + bufferPos);
-      builder.setEndOfStream(false);
-    }
-    return builder.build();
+    throw new IllegalStateException("Illegal method access");
   }
 
   @Override
   public boolean isOpen() {
-    return isOpen;
+    throw new IllegalStateException("Illegal method access");
   }
 
   @Override
   public void close() {
-    if (isOpen) {
-      buffer = null;
-      isOpen = false;
-    }
-  }
-
-  private void validateOpen() throws ClosedChannelException {
-    if (!isOpen) {
-      throw new ClosedChannelException();
-    }
+    throw new IllegalStateException("Illegal method access");
   }
 
   @Override
   public void seek(long position) throws IOException {
-    validateOpen();
-    this.position = position;
-    buffer = null;
-    bufferPos = 0;
-    endOfStream = false;
+    throw new IllegalStateException("Illegal method access");
   }
 
   @Override
   public void setChunkSize(int chunkSize) {
-    this.chunkSize = chunkSize <= 0 ? DEFAULT_CHUNK_SIZE : chunkSize;
+    throw new IllegalStateException("Illegal method access");
   }
 
   @Override
   public int read(ByteBuffer byteBuffer) throws IOException {
-    validateOpen();
-    if (buffer == null) {
-      if (endOfStream) {
-        return -1;
-      }
-      final int toRead =
-          Math.toIntExact(Math.min(limit - position, Math.max(byteBuffer.remaining(), chunkSize)));
-      if (toRead <= 0) {
-        endOfStream = true;
-        return -1;
-      }
-      try {
-        ResultRetryAlgorithm<?> algorithm =
-            retryAlgorithmManager.getForObjectsGet(storageObject, requestOptions);
-        Tuple<String, byte[]> result =
-            runWithRetries(
-                () -> storageRpc.read(storageObject, requestOptions, position, toRead),
-                serviceOptions.getRetrySettings(),
-                algorithm,
-                serviceOptions.getClock());
-        String etag = result.x();
-        byte[] bytes = result.y();
-        if (bytes.length > 0 && lastEtag != null && !Objects.equals(etag, lastEtag)) {
-          throw new IOException("Blob " + blob + " was updated while reading");
-        }
-        lastEtag = etag;
-        buffer = bytes;
-      } catch (RetryHelper.RetryHelperException e) {
-        throw new IOException(e.getCause());
-      }
-      if (toRead > buffer.length) {
-        endOfStream = true;
-        if (buffer.length == 0) {
-          buffer = null;
-          return -1;
-        }
-      }
-    }
-    int toWrite = Math.min(buffer.length - bufferPos, byteBuffer.remaining());
-    byteBuffer.put(buffer, bufferPos, toWrite);
-    bufferPos += toWrite;
-    if (bufferPos >= buffer.length) {
-      position += buffer.length;
-      buffer = null;
-      bufferPos = 0;
-    }
-    return toWrite;
+    throw new IllegalStateException("Illegal method access");
   }
 
   @Override
   public ReadChannel limit(long limit) {
-    Preconditions.checkArgument(limit >= 0, "Limit must be >= 0");
-    this.limit = limit;
-    return this;
+    throw new IllegalStateException("Illegal method access");
   }
 
   @Override
   public long limit() {
-    return limit;
+    throw new IllegalStateException("Illegal method access");
   }
 
+  /** Retained for binary compatibility. Will be removed at next major version! */
+  @SuppressWarnings("unused")
+  @Deprecated
+  @VisibleForTesting
   static class StateImpl implements RestorableState<ReadChannel>, Serializable {
 
     private static final long serialVersionUID = 7784852608213694645L;
 
-    private final HttpStorageOptions serviceOptions;
-    private final BlobId blob;
-    private final Map<StorageRpc.Option, ?> requestOptions;
-    private final String lastEtag;
-    private final long position;
-    private final boolean isOpen;
-    private final boolean endOfStream;
-    private final int chunkSize;
-    private final long limit;
-
-    StateImpl(Builder builder) {
-      this.serviceOptions = builder.serviceOptions;
-      this.blob = builder.blob;
-      this.requestOptions = builder.requestOptions;
-      this.lastEtag = builder.lastEtag;
-      this.position = builder.position;
-      this.isOpen = builder.isOpen;
-      this.endOfStream = builder.endOfStream;
-      this.chunkSize = builder.chunkSize;
-      this.limit = builder.limit;
-    }
-
-    static class Builder {
-      private final HttpStorageOptions serviceOptions;
-      private final BlobId blob;
-      private final Map<StorageRpc.Option, ?> requestOptions;
-      private String lastEtag;
-      private long position;
-      private boolean isOpen;
-      private boolean endOfStream;
-      private int chunkSize;
-      private long limit;
-
-      private Builder(
-          HttpStorageOptions options, BlobId blob, Map<StorageRpc.Option, ?> reqOptions) {
-        this.serviceOptions = options;
-        this.blob = blob;
-        this.requestOptions = reqOptions;
-      }
-
-      Builder setLastEtag(String lastEtag) {
-        this.lastEtag = lastEtag;
-        return this;
-      }
-
-      Builder setPosition(long position) {
-        this.position = position;
-        return this;
-      }
-
-      Builder setIsOpen(boolean isOpen) {
-        this.isOpen = isOpen;
-        return this;
-      }
-
-      Builder setEndOfStream(boolean endOfStream) {
-        this.endOfStream = endOfStream;
-        return this;
-      }
-
-      Builder setChunkSize(int chunkSize) {
-        this.chunkSize = chunkSize;
-        return this;
-      }
-
-      Builder setLimit(long limit) {
-        this.limit = limit;
-        return this;
-      }
+    // the following fields are dangling as they are only set via object deserialization, and only
+    // read in #restore()
+    private HttpStorageOptions serviceOptions;
+    private BlobId blob;
+    private Map<StorageRpc.Option, ?> requestOptions;
+    private String lastEtag;
+    private long position;
+    private boolean isOpen;
+    private boolean endOfStream;
+    private int chunkSize;
+    private long limit;
 
-      RestorableState<ReadChannel> build() {
-        return new StateImpl(this);
-      }
-    }
-
-    static Builder builder(
-        HttpStorageOptions options, BlobId blob, Map<StorageRpc.Option, ?> reqOptions) {
-      return new Builder(options, blob, reqOptions);
-    }
+    private StateImpl() {}
 
     @Override
     public ReadChannel restore() {
-      BlobReadChannel channel = new BlobReadChannel(serviceOptions, blob, requestOptions);
-      channel.lastEtag = lastEtag;
-      channel.position = position;
-      channel.isOpen = isOpen;
-      channel.endOfStream = endOfStream;
-      channel.chunkSize = chunkSize;
-      channel.limit = limit;
+      StorageObject encode = Conversions.apiary().blobId().encode(blob);
+      BlobReadChannelV2 channel =
+          new BlobReadChannelV2(
+              encode, requestOptions, BlobReadChannelContext.from(serviceOptions));
+      try {
+        channel.seek(position);
+        channel.limit(limit);
+        channel.setChunkSize(chunkSize);
+        if (!isOpen) {
+          channel.close();
+        }
+      } catch (IOException e) {
+        throw StorageException.coalesce(e);
+      }
       return channel;
     }