chore(storage): remove the fast open prototype

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

StorageDataClient.readSession(req, ctx, fastOpenConfigs) and the session's
fast open read now do what the single-range prototype did, so remove
fastOpenReadSession, FastOpenObjectReadSession and the Borrowable lease on
ObjectReadSessionStream, which only the prototype used.

The prototype's tests move to the real path: they open with readSession and
read with readAs, keeping fast open coverage for byte array, ByteString and
channel reads.

Author: Dhriti Chopra

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

@@ -31,7 +31,6 @@
 import com.google.cloud.storage.ObjectReadSessionState.OpenArguments;
 import com.google.cloud.storage.ResponseContentLifecycleHandle.ChildRef;
 import com.google.cloud.storage.RetryContext.OnSuccess;
-import com.google.cloud.storage.StorageDataClient.Borrowable;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import com.google.protobuf.ByteString;
@@ -58,7 +57,7 @@
 import org.checkerframework.checker.nullness.qual.Nullable;
 
 final class ObjectReadSessionStream
-    implements ClientStream<BidiReadObjectRequest>, ApiFuture<Void>, IOAutoCloseable, Borrowable {
+    implements ClientStream<BidiReadObjectRequest>, ApiFuture<Void>, IOAutoCloseable {
 
   private final SettableApiFuture<Void> objectReadSessionResolveFuture;
 
@@ -206,11 +205,6 @@ boolean isOpen() {
     return openLeases.get() > 0;
   }
 
-  public void borrow() {
-    checkOpen();
-    openLeases.incrementAndGet();
-  }
-
   private void checkOpen() {
     Preconditions.checkState(isOpen(), "Stream closed");
   }

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

@@ -23,14 +23,11 @@
 import com.google.api.core.InternalApi;
 import com.google.api.gax.grpc.GrpcCallContext;
 import com.google.cloud.storage.GrpcUtils.ZeroCopyBidiStreamingCallable;
-import com.google.cloud.storage.ObjectReadSessionState.OpenArguments;
 import com.google.cloud.storage.ReadProjectionConfig.ProjectionType;
 import com.google.cloud.storage.RetryContext.RetryContextProvider;
 import com.google.common.collect.ImmutableList;
-import com.google.common.collect.ImmutableMap;
 import com.google.storage.v2.BidiReadObjectRequest;
 import com.google.storage.v2.BidiReadObjectResponse;
-import java.io.IOException;
 import java.time.Duration;
 import java.util.HashMap;
 import java.util.List;
@@ -117,46 +114,6 @@ ApiFuture<ObjectReadSession> readSession(
     return objectReadSessionFuture;
   }
 
-  <Projection> ApiFuture<FastOpenObjectReadSession<Projection>> fastOpenReadSession(
-      BidiReadObjectRequest openRequest,
-      GrpcCallContext ctx,
-      ReadProjectionConfig<Projection> config) {
-    checkArgument(
-        openRequest.getReadRangesList().isEmpty(),
-        "ranges included in the initial request are not supported");
-    checkArgument(
-        config.getType() == ProjectionType.STREAM_READ,
-        "unsupported ReadProjectionConfig: %s",
-        config.getClass().getName());
-    ObjectReadSessionState state = new ObjectReadSessionState(ctx, openRequest);
-
-    ZeroCopyBidiStreamingCallable<BidiReadObjectRequest, BidiReadObjectResponse> callable =
-        getCallable();
-    ObjectReadSessionStream stream =
-        ObjectReadSessionStream.create(executor, callable, state, retryContextProvider.create());
-
-    long readId = state.newReadId();
-    ObjectReadSessionStreamRead<Projection> read =
-        config.cast().newRead(readId, retryContextProvider.create());
-    state.putOutstandingRead(readId, read);
-
-    ApiFuture<FastOpenObjectReadSession<Projection>> objectReadSessionFuture =
-        ApiFutures.transform(
-            stream,
-            nowOpen ->
-                new FastOpenObjectReadSession<>(
-                    new ObjectReadSessionImpl(
-                        executor, callable, stream, state, retryContextProvider, ImmutableMap.of()),
-                    read,
-                    stream),
-            executor);
-    OpenArguments openArguments = state.getOpenArguments();
-    BidiReadObjectRequest req = openArguments.getReq();
-    stream.send(req);
-    read.setOnCloseCallback(stream);
-    return objectReadSessionFuture;
-  }
-
   @SuppressWarnings("ResultOfMethodCallIgnored")
   @Override
   public void close() throws Exception {
@@ -182,58 +139,4 @@ static StorageDataClient create(
     return new StorageDataClient(
         executor, terminationAwaitDuration, read, retryContextProvider, onClose);
   }
-
-  @FunctionalInterface
-  interface Borrowable {
-    void borrow();
-  }
-
-  static final class FastOpenObjectReadSession<Projection> implements IOAutoCloseable {
-    private final ObjectReadSession session;
-    private final ObjectReadSessionStreamRead<Projection> read;
-    private final Borrowable borrowable;
-    private boolean sessionLeased;
-
-    private FastOpenObjectReadSession(
-        ObjectReadSession session,
-        ObjectReadSessionStreamRead<Projection> read,
-        Borrowable borrowable) {
-      this.session = session;
-      this.read = read;
-      this.borrowable = borrowable;
-      this.sessionLeased = false;
-    }
-
-    ObjectReadSession getSession() {
-      if (!sessionLeased) {
-        sessionLeased = true;
-        borrowable.borrow();
-      }
-      return session;
-    }
-
-    ObjectReadSessionStreamRead<Projection> getRead() {
-      return read;
-    }
-
-    Projection getProjection() {
-      return read.project();
-    }
-
-    @Override
-    public void close() throws IOException {
-      //noinspection EmptyTryBlock
-      try (IOAutoCloseable ignore1 = session;
-          IOAutoCloseable ignore2 = read) {
-        // use try-with to ensure full cleanup
-      }
-    }
-
-    public static <Projection> FastOpenObjectReadSession<Projection> of(
-        ObjectReadSession session,
-        ObjectReadSessionStreamRead<Projection> read,
-        Borrowable borrowable) {
-      return new FastOpenObjectReadSession<>(session, read, borrowable);
-    }
-  }
 }

java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITObjectReadSessionFakeTest.java

@@ -41,7 +41,6 @@
 import com.google.cloud.storage.Hasher.UncheckedChecksumMismatchException;
 import com.google.cloud.storage.OtelStorageDecorator.OtelDecoratingBlobReadSession;
 import com.google.cloud.storage.Storage.BlobSourceOption;
-import com.google.cloud.storage.StorageDataClient.FastOpenObjectReadSession;
 import com.google.cloud.storage.ZeroCopySupport.DisposableByteString;
 import com.google.cloud.storage.it.ChecksummedTestContent;
 import com.google.cloud.storage.it.GrpcPlainRequestLoggingInterceptor;
@@ -1310,7 +1309,7 @@ public void onCompleted() {
   }
 
   @Test
-  public void gettingSessionFromFastOpenKeepsTheSessionOpenUntilClosed() throws Exception {
+  public void fastOpenRead_thenMoreReadsOnTheSameSession() throws Exception {
     ChecksummedTestContent expected = ChecksummedTestContent.of(ALL_OBJECT_BYTES, 10, 30);
 
     ChecksummedTestContent content1 = ChecksummedTestContent.of(ALL_OBJECT_BYTES, 10, 10);
@@ -1384,19 +1383,17 @@ public void gettingSessionFromFastOpenKeepsTheSessionOpenUntilClosed() throws Ex
                       .build())
               .build();
 
-      ApiFuture<FastOpenObjectReadSession<ApiFuture<DisposableByteString>>> future =
-          dataClient.fastOpenReadSession(
-              req,
-              GrpcCallContext.createDefault(),
-              ReadProjectionConfigs.asFutureByteString().withRangeSpec(RangeSpec.of(10, 10)));
+      ReadAsFutureByteString fastOpenConfig =
+          ReadProjectionConfigs.asFutureByteString().withRangeSpec(RangeSpec.of(10, 10));
+      ApiFuture<ObjectReadSession> future =
+          dataClient.readSession(
+              req, GrpcCallContext.createDefault(), ImmutableList.of(fastOpenConfig));
 
       ByteString bytes = ByteString.empty();
       Exception caught = null;
 
-      try (FastOpenObjectReadSession<ApiFuture<DisposableByteString>> fastOpenChannel =
-              future.get(5, TimeUnit.SECONDS);
-          ObjectReadSession session = fastOpenChannel.getSession()) {
-        ApiFuture<DisposableByteString> futureBytes1 = fastOpenChannel.getProjection();
+      try (ObjectReadSession session = future.get(5, TimeUnit.SECONDS)) {
+        ApiFuture<DisposableByteString> futureBytes1 = session.readAs(fastOpenConfig);
         try (DisposableByteString disposableByteString = futureBytes1.get()) {
           bytes = bytes.concat(disposableByteString.byteString());
         }

java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITStorageDataClientFakeTest.java

@@ -24,9 +24,9 @@
 import com.google.api.core.ApiFuture;
 import com.google.api.gax.grpc.GrpcCallContext;
 import com.google.cloud.storage.ITObjectReadSessionFakeTest.FakeStorage;
-import com.google.cloud.storage.StorageDataClient.FastOpenObjectReadSession;
 import com.google.cloud.storage.ZeroCopySupport.DisposableByteString;
 import com.google.cloud.storage.it.ChecksummedTestContent;
+import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.io.ByteStreams;
 import com.google.protobuf.ByteString;
@@ -137,10 +137,10 @@ private <P> void doTest(ReadProjectionConfig<P> config, ThrowableFunction<P, byt
                       .build())
               .build();
 
-      ApiFuture<FastOpenObjectReadSession<P>> f =
-          dataClient.fastOpenReadSession(req, GrpcCallContext.createDefault(), config);
-      try (FastOpenObjectReadSession<P> fastOpen = f.get(3, TimeUnit.SECONDS)) {
-        byte[] apply = func.apply(fastOpen.getProjection());
+      ApiFuture<ObjectReadSession> f =
+          dataClient.readSession(req, GrpcCallContext.createDefault(), ImmutableList.of(config));
+      try (ObjectReadSession session = f.get(3, TimeUnit.SECONDS)) {
+        byte[] apply = func.apply(session.readAs(config));
         assertThat(xxd(apply)).isEqualTo(xxd(content.getBytes()));
       }
     }