fix: update BidiBlobWriteSessionConfig to respect a provided bufferSi…

…ze (#2471) Refactor, reuse DecoratedWritableByteChannelSession and LazySession from DefaultBlobWriteSessionConfig rather than duplicating in BidiBlobWriteSessionConfig.
googleapis · Mar 26, 2024 · e1fb857 · e1fb857
1 parent 55ec75f
commit e1fb857
Show file tree

Hide file tree

Showing 2 changed files with 21 additions and 62 deletions.
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/...e-cloud-storage/src/main/java/com/google/cloud/storage/DefaultBlobWriteSessionConfig.java b/...e-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;
     }