chore(storage): exposed fastOpenBidiRead to Storage interface.

Author: ShreyasSinha

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

@@ -18,16 +18,29 @@
 
 import com.google.api.core.ApiFuture;
 import com.google.api.core.ApiFutures;
+import com.google.cloud.storage.StorageDataClient.FastOpenObjectReadSession;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.util.concurrent.MoreExecutors;
 import java.io.IOException;
+import java.util.Objects;
 
 final class BlobReadSessionAdapter implements BlobReadSession {
 
   @VisibleForTesting final ObjectReadSession session;
+  private ObjectReadSessionStreamRead<?> fastOpenRead;
+  private final ReadProjectionConfig<?> fastOpenConfig;
 
   BlobReadSessionAdapter(ObjectReadSession session) {
+    this(session, null, null);
+  }
+
+  private BlobReadSessionAdapter(
+      ObjectReadSession session,
+      ObjectReadSessionStreamRead<?> fastOpenRead,
+      ReadProjectionConfig<?> fastOpenConfig) {
     this.session = session;
+    this.fastOpenRead = fastOpenRead;
+    this.fastOpenConfig = fastOpenConfig;
   }
 
   @Override
@@ -40,6 +53,17 @@ public BlobInfo getBlobInfo() {
   @SuppressWarnings({"rawtypes", "unchecked"})
   @Override
   public <Projection> Projection readAs(ReadProjectionConfig<Projection> config) {
+    synchronized (this) {
+      if (fastOpenRead != null && Objects.equals(config, fastOpenConfig)) {
+        Projection projection = (Projection) fastOpenRead.project();
+        fastOpenRead = null;
+        if (projection instanceof ApiFuture) {
+          ApiFuture apiFuture = (ApiFuture) projection;
+          return (Projection) StorageException.coalesceAsync(apiFuture);
+        }
+        return projection;
+      }
+    }
     Projection projection = session.readAs(config);
     if (projection instanceof ApiFuture) {
       ApiFuture apiFuture = (ApiFuture) projection;
@@ -59,4 +83,14 @@ static ApiFuture<BlobReadSession> wrap(ApiFuture<ObjectReadSession> session) {
         BlobReadSessionAdapter::new,
         MoreExecutors.directExecutor());
   }
+
+  static <Projection> ApiFuture<BlobReadSession> wrap(
+      ApiFuture<FastOpenObjectReadSession<Projection>> session,
+      ReadProjectionConfig<Projection> config) {
+    return ApiFutures.transform(
+        StorageException.coalesceAsync(session),
+        fastOpen ->
+            new BlobReadSessionAdapter(fastOpen.getSession(), fastOpen.getRead(), config),
+        MoreExecutors.directExecutor());
+  }
 }

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

@@ -1487,6 +1487,31 @@ public ApiFuture<BlobReadSession> blobReadSession(BlobId id, BlobSourceOption...
     return BlobReadSessionAdapter.wrap(session);
   }
 
+  @Override
+  public <Projection> ApiFuture<BlobReadSession> blobReadSession(
+      BlobId id, ReadProjectionConfig<Projection> config, BlobSourceOption... options) {
+    Opts<ObjectSourceOpt> opts = Opts.unwrap(options);
+    Object object = codecs.blobId().encode(id);
+
+    BidiReadObjectSpec.Builder spec =
+        BidiReadObjectSpec.newBuilder().setBucket(object.getBucket()).setObject(object.getName());
+
+    long generation = object.getGeneration();
+    if (generation > 0) {
+      spec.setGeneration(generation);
+    }
+    BidiReadObjectRequest.Builder b = BidiReadObjectRequest.newBuilder();
+    b.setReadObjectSpec(spec);
+    opts.bidiReadObjectRequest().apply(b);
+    BidiReadObjectRequest req = b.build();
+
+    GrpcCallContext context = opts.grpcMetadataMapper().apply(GrpcCallContext.createDefault());
+    ApiFuture<StorageDataClient.FastOpenObjectReadSession<Projection>> session =
+        storageDataClient.fastOpenReadSession(req, context, config);
+
+    return BlobReadSessionAdapter.wrap(session, config);
+  }
+
   @Override
   public GrpcStorageOptions getOptions() {
     return (GrpcStorageOptions) super.getOptions();

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

@@ -1504,6 +1504,49 @@ public ApiFuture<BlobReadSession> blobReadSession(BlobId id, BlobSourceOption...
     }
   }
 
+  @Override
+  public <Projection> ApiFuture<BlobReadSession> blobReadSession(
+      BlobId id, ReadProjectionConfig<Projection> config, BlobSourceOption... options) {
+    Span blobReadSessionSpan =
+        tracer
+            .spanBuilder(BLOB_READ_SESSION)
+            .setAttribute("gsutil.uri", id.toGsUtilUriWithGeneration())
+            .startSpan();
+    try (Scope ignore1 = blobReadSessionSpan.makeCurrent()) {
+      Context blobReadSessionContext = Context.current();
+      Span ready = tracer.spanBuilder(BLOB_READ_SESSION + "/ready").startSpan();
+      ApiFuture<BlobReadSession> blobReadSessionApiFuture =
+          delegate.blobReadSession(id, config, options);
+      ApiFuture<BlobReadSession> futureDecorated =
+          ApiFutures.transform(
+              blobReadSessionApiFuture,
+              delegate -> {
+                ready.end();
+                return new OtelDecoratingBlobReadSession(
+                    delegate, id, blobReadSessionContext, blobReadSessionSpan);
+              },
+              MoreExecutors.directExecutor());
+      ApiFutures.addCallback(
+          futureDecorated,
+          (OnFailureApiFutureCallback<BlobReadSession>)
+              t -> {
+                blobReadSessionSpan.recordException(t);
+                blobReadSessionSpan.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+                blobReadSessionSpan.end();
+                ready.recordException(t);
+                ready.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+                ready.end();
+              },
+          MoreExecutors.directExecutor());
+      return futureDecorated;
+    } catch (Throwable t) {
+      blobReadSessionSpan.recordException(t);
+      blobReadSessionSpan.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      blobReadSessionSpan.end();
+      throw t;
+    }
+  }
+
   @Override
   public BlobAppendableUpload blobAppendableUpload(
       BlobInfo blobInfo, BlobAppendableUploadConfig uploadConfig, BlobWriteOption... options) {

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

@@ -5955,6 +5955,27 @@ default ApiFuture<BlobReadSession> blobReadSession(BlobId id, BlobSourceOption..
     return throwGrpcOnly(fmtMethodName("blobReadSession", BlobId.class, BlobSourceOption.class));
   }
 
+  /**
+   * Asynchronously set up a new {@link BlobReadSession} for the specified {@link BlobId}, {@link
+   * ReadProjectionConfig} and {@code options}.
+   *
+   * <p>This is a "fast-open" version of {@link #blobReadSession(BlobId, BlobSourceOption...)} which
+   * allows the initial read to be started at the same time as the session is opened.
+   *
+   * @param id the blob to read from
+   * @param config the configuration for the initial read
+   * @param options blob source options
+   * @since 2.65.0 This new api is in preview and is subject to breaking changes.
+   */
+  @BetaApi
+  @TransportCompatibility({Transport.GRPC})
+  default <Projection> ApiFuture<BlobReadSession> blobReadSession(
+      BlobId id, ReadProjectionConfig<Projection> config, BlobSourceOption... options) {
+    return throwGrpcOnly(
+        fmtMethodName(
+            "blobReadSession", BlobId.class, ReadProjectionConfig.class, BlobSourceOption.class));
+  }
+
   /**
    * Create a new {@link BlobAppendableUpload} for the specified {@code blobInfo} and {@code
    * options}.