feat(storage): adding disabling option for bidi reads

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

@@ -151,7 +151,7 @@ private AccumulatingRead(
       super(rangeSpec, retryContext, onCloseCallback);
       this.readId = readId;
       this.hasher =
-          (rangeSpec.begin() == 0 && !(hasher instanceof Hasher.NoOpHasher))
+          (rangeSpec.begin() == 0)
               ? new CumulativeHasher(hasher, 0, rangeSpec.maxLength())
               : hasher;
       this.complete = SettableApiFuture.create();
@@ -284,7 +284,7 @@ static class StreamingRead extends BaseObjectReadSessionStreamRead<ScatteringByt
       super(rangeSpec, retryContext, onCloseCallback);
       this.readId = new AtomicLong(readId);
       this.hasher =
-          (rangeSpec.begin() == 0 && !(hasher instanceof Hasher.NoOpHasher))
+          (rangeSpec.begin() == 0)
               ? new CumulativeHasher(hasher, 0, rangeSpec.maxLength())
               : hasher;
       this.closed = false;

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

@@ -40,7 +40,7 @@ final class CumulativeHasher implements Hasher {
     this.delegate = delegate;
     this.startOffset = startOffset;
     this.limit = limit;
-    this.cumulativeHash = Crc32cValue.zero();
+    this.cumulativeHash = delegate.initialValue();
   }
 
   @Override
@@ -123,16 +123,19 @@ boolean qualifiesForVerification(Object metadata) {
   void validateCumulativeChecksum(Object metadata)
       throws UncheckedCumulativeChecksumMismatchException {
     if (qualifiesForVerification(metadata)) {
-      Crc32cValue<?> expected = Crc32cValue.of(metadata.getChecksums().getCrc32C());
       Crc32cLengthKnown actual = getCumulativeHash();
+      if (actual == null) {
+        return;
+      }
+      Crc32cValue<?> expected = Crc32cValue.of(metadata.getChecksums().getCrc32C());
       if (!actual.eqValue(expected)) {
         throw new UncheckedCumulativeChecksumMismatchException(expected, actual);
       }
     }
   }
 
   private void accumulate(Crc32cLengthKnown actual) {
-    cumulativeHash = cumulativeHash.concat(actual);
+    cumulativeHash = nullSafeConcat(cumulativeHash, actual);
   }
 
   Crc32cLengthKnown getCumulativeHash() {

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

@@ -67,7 +67,7 @@ private ReadableByteChannelSessionBuilder(
       this.read = read;
       this.retrier = retrier;
       this.resultRetryAlgorithm = resultRetryAlgorithm;
-      this.hasher = Hasher.defaultHasher();
+      this.hasher = Hasher.readHasher();
       this.autoGzipDecompression = false;
     }
 

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

@@ -93,7 +93,7 @@ final class GapicUnbufferedReadableByteChannel
     this.read = read;
     this.req = req;
     this.hasher =
-        (req.getReadOffset() == 0 && !(hasher instanceof Hasher.NoOpHasher))
+        (req.getReadOffset() == 0)
             ? new CumulativeHasher(
                 hasher,
                 0,

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

@@ -62,7 +62,7 @@ protected LazyReadChannel<?, Object> newLazyReadChannel() {
               ResumableMedia.gapic()
                   .read()
                   .byteChannel(read, retrier, resultRetryAlgorithm)
-                  .setHasher(Hasher.defaultHasher())
+                  .setHasher(Hasher.readHasher())
                   .setAutoGzipDecompression(autoGzipDecompression);
           BufferHandle bufferHandle = getBufferHandle();
           // because we're erasing the specific type of channel, we need to declare it here.

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

@@ -57,6 +57,27 @@ final class DefaultInstanceHolder {
     }
   }
 
+  final class ReadInstanceHolder {
+    private static final Logger LOGGER = Logger.getLogger(Hasher.class.getName());
+    private static final String PROPERTY_NAME = "com.google.cloud.storage.Hasher.read";
+    private static final String PROPERTY_VALUE =
+        System.getProperty(PROPERTY_NAME, DefaultInstanceHolder.PROPERTY_VALUE);
+    static final Hasher READ_HASHER;
+
+    static {
+      LOGGER.fine(String.format(Locale.US, "-D%s=%s", PROPERTY_NAME, PROPERTY_VALUE));
+      if ("disabled".equalsIgnoreCase(PROPERTY_VALUE)) {
+        READ_HASHER = noop();
+      } else {
+        READ_HASHER = enabled();
+      }
+    }
+  }
+
+  static Hasher readHasher() {
+    return ReadInstanceHolder.READ_HASHER;
+  }
+
   @Nullable
   default Crc32cLengthKnown hash(Supplier<ByteBuffer> b) {
     return hash(b.get());

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

@@ -46,7 +46,7 @@
 @BetaApi
 @Immutable
 public final class ReadAsChannel extends BaseConfig<ScatteringByteChannel, StreamingRead> {
-  static final ReadAsChannel INSTANCE = new ReadAsChannel(RangeSpec.all(), Hasher.enabled());
+  static final ReadAsChannel INSTANCE = new ReadAsChannel(RangeSpec.all(), Hasher.readHasher());
 
   private final RangeSpec range;
   private final Hasher hasher;
@@ -97,7 +97,7 @@ public ReadAsChannel withRangeSpec(RangeSpec range) {
    */
   @BetaApi
   boolean getCrc32cValidationEnabled() {
-    return Hasher.enabled().equals(hasher);
+    return !Hasher.noop().equals(hasher);
   }
 
   /**
@@ -111,7 +111,7 @@ boolean getCrc32cValidationEnabled() {
    */
   @BetaApi
   ReadAsChannel withCrc32cValidationEnabled(boolean enabled) {
-    if (enabled && Hasher.enabled().equals(hasher)) {
+    if (enabled && !Hasher.noop().equals(hasher)) {
       return this;
     } else if (!enabled && Hasher.noop().equals(hasher)) {
       return this;

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

@@ -46,7 +46,7 @@ public final class ReadAsFutureByteString
     extends BaseConfig<ApiFuture<DisposableByteString>, AccumulatingRead<DisposableByteString>> {
 
   static final ReadAsFutureByteString INSTANCE =
-      new ReadAsFutureByteString(RangeSpec.all(), Hasher.enabled());
+      new ReadAsFutureByteString(RangeSpec.all(), Hasher.readHasher());
 
   private final RangeSpec range;
   private final Hasher hasher;
@@ -97,7 +97,7 @@ public ReadAsFutureByteString withRangeSpec(RangeSpec range) {
    */
   @BetaApi
   boolean getCrc32cValidationEnabled() {
-    return Hasher.enabled().equals(hasher);
+    return !Hasher.noop().equals(hasher);
   }
 
   /**
@@ -111,7 +111,7 @@ boolean getCrc32cValidationEnabled() {
    */
   @BetaApi
   ReadAsFutureByteString withCrc32cValidationEnabled(boolean enabled) {
-    if (enabled && Hasher.enabled().equals(hasher)) {
+    if (enabled && !Hasher.noop().equals(hasher)) {
       return this;
     } else if (!enabled && Hasher.noop().equals(hasher)) {
       return this;

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

@@ -41,7 +41,7 @@ public final class ReadAsFutureBytes
     extends BaseConfig<ApiFuture<byte[]>, AccumulatingRead<byte[]>> {
 
   static final ReadAsFutureBytes INSTANCE =
-      new ReadAsFutureBytes(RangeSpec.all(), Hasher.enabled());
+      new ReadAsFutureBytes(RangeSpec.all(), Hasher.readHasher());
 
   private final RangeSpec range;
   private final Hasher hasher;
@@ -92,7 +92,7 @@ public ReadAsFutureBytes withRangeSpec(RangeSpec range) {
    */
   @BetaApi
   boolean getCrc32cValidationEnabled() {
-    return Hasher.enabled().equals(hasher);
+    return !Hasher.noop().equals(hasher);
   }
 
   /**
@@ -106,7 +106,7 @@ boolean getCrc32cValidationEnabled() {
    */
   @BetaApi
   ReadAsFutureBytes withCrc32cValidationEnabled(boolean enabled) {
-    if (enabled && Hasher.enabled().equals(hasher)) {
+    if (enabled && !Hasher.noop().equals(hasher)) {
       return this;
     } else if (!enabled && Hasher.noop().equals(hasher)) {
       return this;

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

@@ -46,7 +46,8 @@
 public final class ReadAsSeekableChannel extends ReadProjectionConfig<SeekableByteChannel> {
 
   static final ReadAsSeekableChannel INSTANCE =
-      new ReadAsSeekableChannel(Hasher.enabled(), LinearExponentialRangeSpecFunction.INSTANCE);
+      new ReadAsSeekableChannel(
+          Hasher.readHasher(), LinearExponentialRangeSpecFunction.INSTANCE);
 
   private final Hasher hasher;
   private final RangeSpecFunction rangeSpecFunction;
@@ -94,7 +95,7 @@ public ReadAsSeekableChannel withRangeSpecFunction(RangeSpecFunction rangeSpecFu
    */
   @BetaApi
   boolean getCrc32cValidationEnabled() {
-    return Hasher.enabled().equals(hasher);
+    return !Hasher.noop().equals(hasher);
   }
 
   /**
@@ -108,7 +109,7 @@ boolean getCrc32cValidationEnabled() {
    */
   @BetaApi
   ReadAsSeekableChannel withCrc32cValidationEnabled(boolean enabled) {
-    if (enabled && Hasher.enabled().equals(hasher)) {
+    if (enabled && !Hasher.noop().equals(hasher)) {
       return this;
     } else if (!enabled && Hasher.noop().equals(hasher)) {
       return this;