fix: update BidiBlobWriteSessionConfig to respect a provided bufferSize (#2471)

BenWhitehead · web-flow · commit e1fb8577e6b7 · 2024-03-26T18:25:12.000-04:00
Refactor, reuse DecoratedWritableByteChannelSession and LazySession from DefaultBlobWriteSessionConfig rather than duplicating in BidiBlobWriteSessionConfig.
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/BidiBlobWriteSessionConfig.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/BidiBlobWriteSessionConfig.java
@@ -17,19 +17,31 @@
 package com.google.cloud.storage;
 
 import com.google.api.core.ApiFuture;
-import com.google.api.core.ApiFutures;
 import com.google.api.core.BetaApi;
 import com.google.api.core.InternalApi;
 import com.google.api.gax.grpc.GrpcCallContext;
+import com.google.cloud.storage.DefaultBlobWriteSessionConfig.DecoratedWritableByteChannelSession;
+import com.google.cloud.storage.DefaultBlobWriteSessionConfig.LazySession;
+import com.google.cloud.storage.TransportCompatibility.Transport;
 import com.google.common.base.Preconditions;
-import com.google.common.util.concurrent.MoreExecutors;
 import com.google.storage.v2.BidiWriteObjectRequest;
 import com.google.storage.v2.BidiWriteObjectResponse;
 import java.io.IOException;
-import java.nio.channels.WritableByteChannel;
 import java.time.Clock;
+import javax.annotation.concurrent.Immutable;
 
-public class BidiBlobWriteSessionConfig extends BlobWriteSessionConfig
+/**
+ * Perform a resumable upload, uploading at most {@code bufferSize} bytes each flush.
+ *
+ * <p>Configuration of buffer size can be performed via {@link
+ * BidiBlobWriteSessionConfig#withBufferSize(int)}.
+ *
+ * @since 2.34.0 This new api is in preview and is subject to breaking changes.
+ */
+@Immutable
+@BetaApi
+@TransportCompatibility({Transport.GRPC})
+public final class BidiBlobWriteSessionConfig extends BlobWriteSessionConfig
     implements BlobWriteSessionConfig.GrpcCompatible {
   private static final long serialVersionUID = -903533790705476197L;
 
@@ -54,7 +66,7 @@ public int getBufferSize() {
 
   @Override
   WriterFactory createFactory(Clock clock) throws IOException {
-    return new Factory(ByteSizeConstants._16MiB);
+    return new Factory(bufferSize);
   }
 
   @InternalApi
@@ -124,57 +136,4 @@ public BidiBlobWriteSessionConfig withBufferSize(int bufferSize) {
         ByteSizeConstants._256KiB);
     return new BidiBlobWriteSessionConfig(bufferSize);
   }
-
-  private static final class DecoratedWritableByteChannelSession<WBC extends WritableByteChannel, T>
-      implements WritableByteChannelSession<WBC, BlobInfo> {
-
-    private final WritableByteChannelSession<WBC, T> delegate;
-    private final Conversions.Decoder<T, BlobInfo> decoder;
-
-    private DecoratedWritableByteChannelSession(
-        WritableByteChannelSession<WBC, T> delegate, Conversions.Decoder<T, BlobInfo> decoder) {
-      this.delegate = delegate;
-      this.decoder = decoder;
-    }
-
-    @Override
-    public WBC open() {
-      try {
-        return WritableByteChannelSession.super.open();
-      } catch (Exception e) {
-        throw StorageException.coalesce(e);
-      }
-    }
-
-    @Override
-    public ApiFuture<WBC> openAsync() {
-      return delegate.openAsync();
-    }
-
-    @Override
-    public ApiFuture<BlobInfo> getResult() {
-      return ApiFutures.transform(
-          delegate.getResult(), decoder::decode, MoreExecutors.directExecutor());
-    }
-  }
-
-  private static final class LazySession<R>
-      implements WritableByteChannelSession<
-          BufferedWritableByteChannelSession.BufferedWritableByteChannel, R> {
-    private final LazyWriteChannel<R> lazy;
-
-    private LazySession(LazyWriteChannel<R> lazy) {
-      this.lazy = lazy;
-    }
-
-    @Override
-    public ApiFuture<BufferedWritableByteChannelSession.BufferedWritableByteChannel> openAsync() {
-      return lazy.getSession().openAsync();
-    }
-
-    @Override
-    public ApiFuture<R> getResult() {
-      return lazy.getSession().getResult();
-    }
-  }
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/DefaultBlobWriteSessionConfig.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/DefaultBlobWriteSessionConfig.java
@@ -153,13 +153,13 @@ public WritableByteChannelSession<?, BlobInfo> writeSession(
     }
   }
 
-  private static final class DecoratedWritableByteChannelSession<WBC extends WritableByteChannel, T>
+  static final class DecoratedWritableByteChannelSession<WBC extends WritableByteChannel, T>
       implements WritableByteChannelSession<WBC, BlobInfo> {
 
     private final WritableByteChannelSession<WBC, T> delegate;
     private final Decoder<T, BlobInfo> decoder;
 
-    private DecoratedWritableByteChannelSession(
+    DecoratedWritableByteChannelSession(
         WritableByteChannelSession<WBC, T> delegate, Decoder<T, BlobInfo> decoder) {
       this.delegate = delegate;
       this.decoder = decoder;
@@ -186,11 +186,11 @@ public ApiFuture<BlobInfo> getResult() {
     }
   }
 
-  private static final class LazySession<R>
+  static final class LazySession<R>
       implements WritableByteChannelSession<BufferedWritableByteChannel, R> {
     private final LazyWriteChannel<R> lazy;
 
-    private LazySession(LazyWriteChannel<R> lazy) {
+    LazySession(LazyWriteChannel<R> lazy) {
       this.lazy = lazy;
     }
 

Original file line number	Diff line number	Diff line change
`@@ -153,13 +153,13 @@ public WritableByteChannelSession<?, BlobInfo> writeSession(`
`153`	`153`	`}`
`154`	`154`	`}`
`155`	`155`
`156`		`- private static final class DecoratedWritableByteChannelSession<WBC extends WritableByteChannel, T>`
	`156`	`+ static final class DecoratedWritableByteChannelSession<WBC extends WritableByteChannel, T>`
`157`	`157`	`implements WritableByteChannelSession<WBC, BlobInfo> {`
`158`	`158`
`159`	`159`	`private final WritableByteChannelSession<WBC, T> delegate;`
`160`	`160`	`private final Decoder<T, BlobInfo> decoder;`
`161`	`161`
`162`		`- private DecoratedWritableByteChannelSession(`
	`162`	`+ DecoratedWritableByteChannelSession(`
`163`	`163`	`WritableByteChannelSession<WBC, T> delegate, Decoder<T, BlobInfo> decoder) {`
`164`	`164`	`this.delegate = delegate;`
`165`	`165`	`this.decoder = decoder;`
`@@ -186,11 +186,11 @@ public ApiFuture<BlobInfo> getResult() {`
`186`	`186`	`}`
`187`	`187`	`}`
`188`	`188`
`189`		`- private static final class LazySession<R>`
	`189`	`+ static final class LazySession<R>`
`190`	`190`	`implements WritableByteChannelSession<BufferedWritableByteChannel, R> {`
`191`	`191`	`private final LazyWriteChannel<R> lazy;`
`192`	`192`
`193`		`- private LazySession(LazyWriteChannel<R> lazy) {`
	`193`	`+ LazySession(LazyWriteChannel<R> lazy) {`
`194`	`194`	`this.lazy = lazy;`
`195`	`195`	`}`
`196`	`196`