Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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);

Expand All @@ -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);
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand All @@ -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,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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}.
Expand Down
Loading
Loading