diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/ObjectReadSessionStream.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/ObjectReadSessionStream.java index beda7e5f7732..d7f33bdcdbe3 100644 --- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/ObjectReadSessionStream.java +++ b/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, ApiFuture, IOAutoCloseable, Borrowable { + implements ClientStream, ApiFuture, IOAutoCloseable { private final SettableApiFuture 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"); } diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageDataClient.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageDataClient.java index a531d55974cf..ec4ed12b2f16 100644 --- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageDataClient.java +++ b/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 readSession( return objectReadSessionFuture; } - ApiFuture> fastOpenReadSession( - BidiReadObjectRequest openRequest, - GrpcCallContext ctx, - ReadProjectionConfig 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 callable = - getCallable(); - ObjectReadSessionStream stream = - ObjectReadSessionStream.create(executor, callable, state, retryContextProvider.create()); - - long readId = state.newReadId(); - ObjectReadSessionStreamRead read = - config.cast().newRead(readId, retryContextProvider.create()); - state.putOutstandingRead(readId, read); - - ApiFuture> 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 implements IOAutoCloseable { - private final ObjectReadSession session; - private final ObjectReadSessionStreamRead read; - private final Borrowable borrowable; - private boolean sessionLeased; - - private FastOpenObjectReadSession( - ObjectReadSession session, - ObjectReadSessionStreamRead 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 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 FastOpenObjectReadSession of( - ObjectReadSession session, - ObjectReadSessionStreamRead read, - Borrowable borrowable) { - return new FastOpenObjectReadSession<>(session, read, borrowable); - } - } } diff --git a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITObjectReadSessionFakeTest.java b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITObjectReadSessionFakeTest.java index 1cb32704982b..634ecfae3c6f 100644 --- a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITObjectReadSessionFakeTest.java +++ b/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>> future = - dataClient.fastOpenReadSession( - req, - GrpcCallContext.createDefault(), - ReadProjectionConfigs.asFutureByteString().withRangeSpec(RangeSpec.of(10, 10))); + ReadAsFutureByteString fastOpenConfig = + ReadProjectionConfigs.asFutureByteString().withRangeSpec(RangeSpec.of(10, 10)); + ApiFuture future = + dataClient.readSession( + req, GrpcCallContext.createDefault(), ImmutableList.of(fastOpenConfig)); ByteString bytes = ByteString.empty(); Exception caught = null; - try (FastOpenObjectReadSession> fastOpenChannel = - future.get(5, TimeUnit.SECONDS); - ObjectReadSession session = fastOpenChannel.getSession()) { - ApiFuture futureBytes1 = fastOpenChannel.getProjection(); + try (ObjectReadSession session = future.get(5, TimeUnit.SECONDS)) { + ApiFuture futureBytes1 = session.readAs(fastOpenConfig); try (DisposableByteString disposableByteString = futureBytes1.get()) { bytes = bytes.concat(disposableByteString.byteString()); } diff --git a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITStorageDataClientFakeTest.java b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITStorageDataClientFakeTest.java index 276889ab44c5..b3fd1b73ed4f 100644 --- a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITStorageDataClientFakeTest.java +++ b/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

void doTest(ReadProjectionConfig

config, ThrowableFunction> f = - dataClient.fastOpenReadSession(req, GrpcCallContext.createDefault(), config); - try (FastOpenObjectReadSession

fastOpen = f.get(3, TimeUnit.SECONDS)) { - byte[] apply = func.apply(fastOpen.getProjection()); + ApiFuture 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())); } }