feat(storage): add blobReadSession overload that takes a fast open read

Seventh step of fast open BiDi reads (go/fast-open-bidi-read, b/552263062).

Storage.blobReadSession(BlobId, ReadProjectionConfig, BlobSourceOption...)
sends the read described by the config in the same request that opens the
session. The first readAs with an equal config is given that read, saving the
round trip a separate first read would take. Only asFutureBytes,
asFutureByteString and asChannel configs are accepted, and a null config only
opens the session.

This is additive: blobReadSession(BlobId, BlobSourceOption...) is unchanged.
OtelStorageDecorator passes the config through, so fast open also works with
OpenTelemetry tracing enabled, and a fast open read that fails is recorded as
an error on its readAs span.

Author: Dhriti Chopra

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

@@ -27,6 +27,7 @@
 import static com.google.cloud.storage.Utils.ifNonNull;
 import static com.google.cloud.storage.Utils.projectNameCodec;
 import static com.google.common.base.MoreObjects.firstNonNull;
+import static com.google.common.base.Preconditions.checkArgument;
 import static java.util.Objects.requireNonNull;
 
 import com.google.api.core.ApiFuture;
@@ -55,6 +56,7 @@
 import com.google.cloud.storage.HmacKey.HmacKeyState;
 import com.google.cloud.storage.PostPolicyV4.PostConditionsV4;
 import com.google.cloud.storage.PostPolicyV4.PostFieldsV4;
+import com.google.cloud.storage.ReadProjectionConfig.ProjectionType;
 import com.google.cloud.storage.Retrying.Retrier;
 import com.google.cloud.storage.Retrying.RetrierWithAlg;
 import com.google.cloud.storage.Storage.ComposeRequest.SourceBlob;
@@ -1468,6 +1470,26 @@ public Blob moveBlob(MoveBlobRequest request) {
 
   @Override
   public ApiFuture<BlobReadSession> blobReadSession(BlobId id, BlobSourceOption... options) {
+    return blobReadSession(id, ImmutableList.of(), options);
+  }
+
+  @Override
+  public <Projection> ApiFuture<BlobReadSession> blobReadSession(
+      BlobId id,
+      @Nullable ReadProjectionConfig<Projection> fastOpenConfig,
+      BlobSourceOption... options) {
+    if (fastOpenConfig == null) {
+      return blobReadSession(id, ImmutableList.of(), options);
+    }
+    checkArgument(
+        fastOpenConfig.getType() == ProjectionType.STREAM_READ,
+        "fastOpenConfig must be asFutureBytes(), asFutureByteString() or asChannel(), but was %s",
+        fastOpenConfig);
+    return blobReadSession(id, ImmutableList.of(fastOpenConfig), options);
+  }
+
+  private ApiFuture<BlobReadSession> blobReadSession(
+      BlobId id, List<ReadProjectionConfig<?>> fastOpenConfigs, BlobSourceOption... options) {
     Opts<ObjectSourceOpt> opts = Opts.unwrap(options);
     Object object = codecs.blobId().encode(id);
 
@@ -1484,7 +1506,8 @@ public ApiFuture<BlobReadSession> blobReadSession(BlobId id, BlobSourceOption...
     BidiReadObjectRequest req = b.build();
 
     GrpcCallContext context = opts.grpcMetadataMapper().apply(GrpcCallContext.createDefault());
-    ApiFuture<ObjectReadSession> session = storageDataClient.readSession(req, context);
+    ApiFuture<ObjectReadSession> session =
+        storageDataClient.readSession(req, context, fastOpenConfigs);
 
     return BlobReadSessionAdapter.wrap(session);
   }

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

@@ -1476,6 +1476,14 @@ public Blob moveBlob(MoveBlobRequest request) {
 
   @Override
   public ApiFuture<BlobReadSession> blobReadSession(BlobId id, BlobSourceOption... options) {
+    return blobReadSession(id, null, options);
+  }
+
+  @Override
+  public <Projection> ApiFuture<BlobReadSession> blobReadSession(
+      BlobId id,
+      @Nullable ReadProjectionConfig<Projection> fastOpenConfig,
+      BlobSourceOption... options) {
     Span blobReadSessionSpan =
         tracer
             .spanBuilder(BLOB_READ_SESSION)
@@ -1484,7 +1492,8 @@ public ApiFuture<BlobReadSession> blobReadSession(BlobId id, BlobSourceOption...
     try (Scope ignore1 = blobReadSessionSpan.makeCurrent()) {
       Context blobReadSessionContext = Context.current();
       Span ready = tracer.spanBuilder(BLOB_READ_SESSION + "/ready").startSpan();
-      ApiFuture<BlobReadSession> blobReadSessionApiFuture = delegate.blobReadSession(id, options);
+      ApiFuture<BlobReadSession> blobReadSessionApiFuture =
+          delegate.blobReadSession(id, fastOpenConfig, options);
       ApiFuture<BlobReadSession> futureDecorated =
           ApiFutures.transform(
               blobReadSessionApiFuture,

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

@@ -5955,6 +5955,44 @@ 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} and {@code
+   * options}, sending the read described by {@code fastOpenConfig} in the same request that opens
+   * the session ("fast open"). This saves the round trip a separate first read would take.
+   *
+   * <p>The first {@link BlobReadSession#readAs(ReadProjectionConfig)} with a config equal to {@code
+   * fastOpenConfig} is given that read; any other {@code readAs} sends a new read. If the fast open
+   * read fails, the error is reported by the returned future or by that first {@code readAs}.
+   *
+   * <h4>Example of reading the first 1 KiB of an object with fast open:</h4>
+   *
+   * <pre>{@code
+   * ReadAsFutureBytes first1KiB =
+   *     ReadProjectionConfigs.asFutureBytes().withRangeSpec(RangeSpec.of(0, 1024));
+   * try (BlobReadSession session =
+   *     storage.blobReadSession(blobId, first1KiB).get(10, TimeUnit.SECONDS)) {
+   *   byte[] bytes = session.readAs(first1KiB).get(10, TimeUnit.SECONDS);
+   * }
+   * }</pre>
+   *
+   * @param id the blob to read from
+   * @param fastOpenConfig the read to send with the open, or {@code null} to only open the session.
+   *     Must be {@link ReadProjectionConfigs#asFutureBytes()}, {@link
+   *     ReadProjectionConfigs#asFutureByteString()} or {@link ReadProjectionConfigs#asChannel()},
+   *     optionally with a range.
+   * @since 2.77.0 This new api is in preview and is subject to breaking changes.
+   */
+  @BetaApi
+  @TransportCompatibility({Transport.GRPC})
+  default <Projection> ApiFuture<BlobReadSession> blobReadSession(
+      BlobId id,
+      @Nullable ReadProjectionConfig<Projection> fastOpenConfig,
+      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}.