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 43fd503a3679..0ef051bb8c4a 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 @@ -26,10 +26,12 @@ 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.storage.v2.BidiReadObjectRequest; import com.google.storage.v2.BidiReadObjectResponse; import java.io.IOException; import java.time.Duration; +import java.util.List; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; @@ -57,6 +59,21 @@ private StorageDataClient( } ApiFuture readSession(BidiReadObjectRequest req, GrpcCallContext ctx) { + return readSession(req, ctx, ImmutableList.of()); + } + + /** + * Open a new {@link ObjectReadSession} using fast open: the ranges in {@code fastOpenConfigs} are + * sent in the same request that opens the session, saving a round trip before the first bytes. + * + * @param fastOpenConfigs the reads to send with the open, for example {@code + * ReadProjectionConfigs.asFutureBytes().withRangeSpec(RangeSpec.of(0, 1024))}; empty to only + * open the session + */ + ApiFuture readSession( + BidiReadObjectRequest req, + GrpcCallContext ctx, + List> fastOpenConfigs) { checkArgument( req.getReadRangesList().isEmpty(), "ranges included in the initial request are not supported"); @@ -67,13 +84,22 @@ ApiFuture readSession(BidiReadObjectRequest req, GrpcCallCont ObjectReadSessionStream stream = ObjectReadSessionStream.create(executor, callable, state, retryContextProvider.create()); + // Add each fast open read to the session state. The open request below is built from the + // state, so these reads are sent as part of the open. + for (ReadProjectionConfig config : fastOpenConfigs) { + long readId = state.newReadId(); + ObjectReadSessionStreamRead read = + config.cast().newRead(readId, retryContextProvider.create()); + state.putOutstandingRead(readId, read); + } + ApiFuture objectReadSessionFuture = ApiFutures.transform( stream, nowOpen -> new ObjectReadSessionImpl(executor, callable, stream, state, retryContextProvider), executor); - stream.send(req); + stream.send(state.getOpenArguments().getReq()); return objectReadSessionFuture; } diff --git a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITFastOpenFakeTest.java b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITFastOpenFakeTest.java new file mode 100644 index 000000000000..ebc753333ae5 --- /dev/null +++ b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITFastOpenFakeTest.java @@ -0,0 +1,141 @@ +/* + * Copyright 2026 Google LLC + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.google.cloud.storage; + +import static com.google.cloud.storage.ByteSizeConstants._2MiB; +import static com.google.cloud.storage.ITObjectReadSessionFakeTest.getReadRange; +import static com.google.cloud.storage.PackagePrivateMethodWorkarounds.maybeGetStorageDataClient; +import static com.google.common.truth.Truth.assertThat; +import static java.util.concurrent.TimeUnit.SECONDS; + +import com.google.api.gax.grpc.GrpcCallContext; +import com.google.cloud.storage.ITObjectReadSessionFakeTest.FakeStorage; +import com.google.cloud.storage.it.ChecksummedTestContent; +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 com.google.storage.v2.BidiReadObjectSpec; +import com.google.storage.v2.BucketName; +import com.google.storage.v2.Object; +import com.google.storage.v2.ObjectRangeData; +import io.grpc.stub.StreamObserver; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.function.Consumer; +import org.junit.Test; + +public final class ITFastOpenFakeTest { + + private static final byte[] ALL_OBJECT_BYTES = DataGenerator.base64Characters().genBytes(64); + private static final Object METADATA = + Object.newBuilder() + .setBucket(BucketName.format("_", "b")) + .setName("o") + .setGeneration(1) + .setSize(_2MiB) + .build(); + private static final GrpcCallContext CTX = GrpcCallContext.createDefault(); + + private static final BidiReadObjectRequest OPEN = + BidiReadObjectRequest.newBuilder() + .setReadObjectSpec( + BidiReadObjectSpec.newBuilder() + .setBucket(METADATA.getBucket()) + .setObject(METADATA.getName()) + .build()) + .build(); + private static final BidiReadObjectRequest OPEN_WITH_READ_OF_10_TO_20 = + OPEN.toBuilder().addReadRanges(getReadRange(1, 10, 10)).build(); + + private static final ReadAsFutureBytes READ_10_TO_20 = + ReadProjectionConfigs.asFutureBytes().withRangeSpec(RangeSpec.of(10, 10)); + + /** Every request the fake server received, in order. */ + private final List requestsReceived = + Collections.synchronizedList(new ArrayList<>()); + + @Test + public void openRequest_includesTheFastOpenRead() throws Exception { + FakeStorage server = + fakeServer(ImmutableMap.of(OPEN_WITH_READ_OF_10_TO_20, metadataAndBytes(1, 10, 10))); + + try (FakeServer fakeServer = FakeServer.of(server); + Storage storage = fakeServer.getGrpcStorageOptions().getService()) { + StorageDataClient dataClient = maybeGetStorageDataClient(storage); + + ObjectReadSession session = + dataClient.readSession(OPEN, CTX, ImmutableList.of(READ_10_TO_20)).get(5, SECONDS); + session.close(); + } + + assertThat(requestsReceived).containsExactly(OPEN_WITH_READ_OF_10_TO_20); + } + + @Test + public void openRequest_withoutFastOpenReads_isUnchanged() throws Exception { + FakeStorage server = fakeServer(ImmutableMap.of(OPEN, metadataOnly())); + + try (FakeServer fakeServer = FakeServer.of(server); + Storage storage = fakeServer.getGrpcStorageOptions().getService()) { + StorageDataClient dataClient = maybeGetStorageDataClient(storage); + + ObjectReadSession session = + dataClient.readSession(OPEN, CTX, ImmutableList.of()).get(5, SECONDS); + session.close(); + } + + assertThat(requestsReceived).containsExactly(OPEN); + } + + /** + * A fake server that answers each request in {@code responses} and adds it to {@link + * #requestsReceived}. Any other request is answered with an error. + */ + private FakeStorage fakeServer(Map responses) { + ImmutableMap.Builder>> + behaviour = ImmutableMap.builder(); + responses.forEach( + (request, response) -> + behaviour.put( + request, + client -> { + requestsReceived.add(request); + client.onNext(response); + })); + return FakeStorage.of(behaviour.buildOrThrow()); + } + + private static BidiReadObjectResponse metadataOnly() { + return BidiReadObjectResponse.newBuilder().setMetadata(METADATA).build(); + } + + /** The response to an open request that has a read of {@code [offset, offset + length)}. */ + private static BidiReadObjectResponse metadataAndBytes(int readId, int offset, int length) { + return BidiReadObjectResponse.newBuilder() + .setMetadata(METADATA) + .addObjectDataRanges( + ObjectRangeData.newBuilder() + .setChecksummedData( + ChecksummedTestContent.of(ALL_OBJECT_BYTES, offset, length).asChecksummedData()) + .setReadRange(getReadRange(readId, offset, length)) + .setRangeEnd(true)) + .build(); + } +}