Skip to content

Commit 47fde85

Browse files
JesseLovelacegcf-owl-bot[bot]sydney-munro
authored
feat: Add Bidi write feature (#2343)
* feat: Add Bidi write feature * 🦉 Updates from OwlBot post-processor See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md * 🦉 Updates from OwlBot post-processor See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md * add in bidi ssb workload * WIP * Add in bidi fields * Add in DefaultBlobWriteSession test * fix copyright * remove bucket creation line * cleanup object created by ssb * 🦉 Updates from OwlBot post-processor See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md * fix up comments --------- Co-authored-by: Owl Bot <gcf-owl-bot[bot]@users.noreply.github.com> Co-authored-by: Sydney Munro <[email protected]>
1 parent 1d2064b commit 47fde85

17 files changed

+1226
-28
lines changed

‎README.md

+3-3
Original file line numberDiff line numberDiff line change
@@ -57,13 +57,13 @@ implementation 'com.google.cloud:google-cloud-storage'
5757
If you are using Gradle without BOM, add this to your dependencies:
5858

5959
```Groovy
60-
implementation 'com.google.cloud:google-cloud-storage:2.33.0'
60+
implementation 'com.google.cloud:google-cloud-storage:2.34.0'
6161
```
6262

6363
If you are using SBT, add this to your dependencies:
6464

6565
```Scala
66-
libraryDependencies += "com.google.cloud" % "google-cloud-storage" % "2.33.0"
66+
libraryDependencies += "com.google.cloud" % "google-cloud-storage" % "2.34.0"
6767
```
6868
<!-- {x-version-update-end} -->
6969

@@ -428,7 +428,7 @@ Java is a registered trademark of Oracle and/or its affiliates.
428428
[kokoro-badge-link-5]: http://storage.googleapis.com/cloud-devrel-public/java/badges/java-storage/java11.html
429429
[stability-image]: https://img.shields.io/badge/stability-stable-green
430430
[maven-version-image]: https://img.shields.io/maven-central/v/com.google.cloud/google-cloud-storage.svg
431-
[maven-version-link]: https://central.sonatype.com/artifact/com.google.cloud/google-cloud-storage/2.33.0
431+
[maven-version-link]: https://central.sonatype.com/artifact/com.google.cloud/google-cloud-storage/2.34.0
432432
[authentication]: https://github.com/googleapis/google-cloud-java#authentication
433433
[auth-scopes]: https://developers.google.com/identity/protocols/oauth2/scopes
434434
[predefined-iam-roles]: https://cloud.google.com/iam/docs/understanding-roles#predefined_roles

‎google-cloud-storage/src/main/java/com/google/cloud/storage/BidiBlobWriteSessionConfig.java

+180
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,180 @@
1+
/*
2+
* Copyright 2024 Google LLC
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package com.google.cloud.storage;
18+
19+
import com.google.api.core.ApiFuture;
20+
import com.google.api.core.ApiFutures;
21+
import com.google.api.core.BetaApi;
22+
import com.google.api.core.InternalApi;
23+
import com.google.api.gax.grpc.GrpcCallContext;
24+
import com.google.common.base.Preconditions;
25+
import com.google.common.util.concurrent.MoreExecutors;
26+
import com.google.storage.v2.BidiWriteObjectRequest;
27+
import com.google.storage.v2.BidiWriteObjectResponse;
28+
import java.io.IOException;
29+
import java.nio.channels.WritableByteChannel;
30+
import java.time.Clock;
31+
32+
public class BidiBlobWriteSessionConfig extends BlobWriteSessionConfig
33+
implements BlobWriteSessionConfig.GrpcCompatible {
34+
private static final long serialVersionUID = -903533790705476197L;
35+
36+
private final int bufferSize;
37+
38+
@InternalApi
39+
BidiBlobWriteSessionConfig(int bufferSize) {
40+
this.bufferSize = bufferSize;
41+
}
42+
43+
/**
44+
* The number of bytes to hold in the buffer before each flush
45+
*
46+
* <p><i>Default:</i> {@code 16777216 (16 MiB)}
47+
*
48+
* @see #withBufferSize(int)
49+
* @since 2.34.0 This new api is in preview and is subject to breaking changes.
50+
*/
51+
public int getBufferSize() {
52+
return bufferSize;
53+
}
54+
55+
@Override
56+
WriterFactory createFactory(Clock clock) throws IOException {
57+
return new Factory(ByteSizeConstants._16MiB);
58+
}
59+
60+
@InternalApi
61+
private static final class Factory implements WriterFactory {
62+
private static final Conversions.Decoder<BidiWriteObjectResponse, BlobInfo>
63+
WRITE_OBJECT_RESPONSE_BLOB_INFO_DECODER =
64+
Conversions.grpc().blobInfo().compose(BidiWriteObjectResponse::getResource);
65+
66+
private final int bufferSize;
67+
68+
private Factory(int bufferSize) {
69+
this.bufferSize = bufferSize;
70+
}
71+
72+
@InternalApi
73+
@Override
74+
public WritableByteChannelSession<?, BlobInfo> writeSession(
75+
StorageInternal s, BlobInfo info, UnifiedOpts.Opts<UnifiedOpts.ObjectTargetOpt> opts) {
76+
if (s instanceof GrpcStorageImpl) {
77+
return new DecoratedWritableByteChannelSession<>(
78+
new LazySession<>(
79+
new LazyWriteChannel<>(
80+
() -> {
81+
GrpcStorageImpl grpc = (GrpcStorageImpl) s;
82+
GrpcCallContext grpcCallContext =
83+
opts.grpcMetadataMapper().apply(GrpcCallContext.createDefault());
84+
BidiWriteObjectRequest req = grpc.getBidiWriteObjectRequest(info, opts);
85+
86+
ApiFuture<BidiResumableWrite> startResumableWrite =
87+
grpc.startResumableWrite(grpcCallContext, req);
88+
return ResumableMedia.gapic()
89+
.write()
90+
.bidiByteChannel(grpc.storageClient.bidiWriteObjectCallable())
91+
.setHasher(Hasher.noop())
92+
.setByteStringStrategy(ByteStringStrategy.copy())
93+
.resumable()
94+
.withRetryConfig(
95+
grpc.getOptions(), grpc.retryAlgorithmManager.idempotent())
96+
.buffered(BufferHandle.allocate(bufferSize))
97+
.setStartAsync(startResumableWrite)
98+
.build();
99+
})),
100+
WRITE_OBJECT_RESPONSE_BLOB_INFO_DECODER);
101+
} else {
102+
throw new IllegalStateException(
103+
"Unknown Storage implementation: " + s.getClass().getName());
104+
}
105+
}
106+
}
107+
108+
/**
109+
* Create a new instance with the {@code bufferSize} set to the specified value.
110+
*
111+
* <p><i>Default:</i> {@code 16777216 (16 MiB)}
112+
*
113+
* @param bufferSize The number of bytes to hold in the buffer before each flush. Must be &gt;=
114+
* {@code 262144 (256 KiB)}
115+
* @return The new instance
116+
* @see #getBufferSize()
117+
* @since 2.34.0 This new api is in preview and is subject to breaking changes.
118+
*/
119+
@BetaApi
120+
public BidiBlobWriteSessionConfig withBufferSize(int bufferSize) {
121+
Preconditions.checkArgument(
122+
bufferSize >= ByteSizeConstants._256KiB,
123+
"bufferSize must be >= %d",
124+
ByteSizeConstants._256KiB);
125+
return new BidiBlobWriteSessionConfig(bufferSize);
126+
}
127+
128+
private static final class DecoratedWritableByteChannelSession<WBC extends WritableByteChannel, T>
129+
implements WritableByteChannelSession<WBC, BlobInfo> {
130+
131+
private final WritableByteChannelSession<WBC, T> delegate;
132+
private final Conversions.Decoder<T, BlobInfo> decoder;
133+
134+
private DecoratedWritableByteChannelSession(
135+
WritableByteChannelSession<WBC, T> delegate, Conversions.Decoder<T, BlobInfo> decoder) {
136+
this.delegate = delegate;
137+
this.decoder = decoder;
138+
}
139+
140+
@Override
141+
public WBC open() {
142+
try {
143+
return WritableByteChannelSession.super.open();
144+
} catch (Exception e) {
145+
throw StorageException.coalesce(e);
146+
}
147+
}
148+
149+
@Override
150+
public ApiFuture<WBC> openAsync() {
151+
return delegate.openAsync();
152+
}
153+
154+
@Override
155+
public ApiFuture<BlobInfo> getResult() {
156+
return ApiFutures.transform(
157+
delegate.getResult(), decoder::decode, MoreExecutors.directExecutor());
158+
}
159+
}
160+
161+
private static final class LazySession<R>
162+
implements WritableByteChannelSession<
163+
BufferedWritableByteChannelSession.BufferedWritableByteChannel, R> {
164+
private final LazyWriteChannel<R> lazy;
165+
166+
private LazySession(LazyWriteChannel<R> lazy) {
167+
this.lazy = lazy;
168+
}
169+
170+
@Override
171+
public ApiFuture<BufferedWritableByteChannelSession.BufferedWritableByteChannel> openAsync() {
172+
return lazy.getSession().openAsync();
173+
}
174+
175+
@Override
176+
public ApiFuture<R> getResult() {
177+
return lazy.getSession().getResult();
178+
}
179+
}
180+
}

‎google-cloud-storage/src/main/java/com/google/cloud/storage/BidiResumableWrite.java

+96
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,96 @@
1+
/*
2+
* Copyright 2023 Google LLC
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package com.google.cloud.storage;
18+
19+
import static com.google.cloud.storage.StorageV2ProtoUtils.fmtProto;
20+
21+
import com.google.cloud.storage.BidiWriteCtx.BidiWriteObjectRequestBuilderFactory;
22+
import com.google.storage.v2.BidiWriteObjectRequest;
23+
import com.google.storage.v2.StartResumableWriteRequest;
24+
import com.google.storage.v2.StartResumableWriteResponse;
25+
import java.util.Objects;
26+
import java.util.function.Function;
27+
import org.checkerframework.checker.nullness.qual.Nullable;
28+
29+
final class BidiResumableWrite implements BidiWriteObjectRequestBuilderFactory {
30+
31+
private final StartResumableWriteRequest req;
32+
private final StartResumableWriteResponse res;
33+
34+
private final BidiWriteObjectRequest writeRequest;
35+
36+
public BidiResumableWrite(
37+
StartResumableWriteRequest req,
38+
StartResumableWriteResponse res,
39+
Function<String, BidiWriteObjectRequest> f) {
40+
this.req = req;
41+
this.res = res;
42+
this.writeRequest = f.apply(res.getUploadId());
43+
}
44+
45+
public StartResumableWriteRequest getReq() {
46+
return req;
47+
}
48+
49+
public StartResumableWriteResponse getRes() {
50+
return res;
51+
}
52+
53+
@Override
54+
public BidiWriteObjectRequest.Builder newBuilder() {
55+
return writeRequest.toBuilder();
56+
}
57+
58+
@Override
59+
public @Nullable String bucketName() {
60+
if (req.hasWriteObjectSpec() && req.getWriteObjectSpec().hasResource()) {
61+
return req.getWriteObjectSpec().getResource().getBucket();
62+
}
63+
return null;
64+
}
65+
66+
@Override
67+
public String toString() {
68+
return "BidiResumableWrite{" + "req=" + fmtProto(req) + ", res=" + fmtProto(res) + '}';
69+
}
70+
71+
@Override
72+
public boolean equals(Object o) {
73+
if (this == o) {
74+
return true;
75+
}
76+
if (!(o instanceof ResumableWrite)) {
77+
return false;
78+
}
79+
ResumableWrite resumableWrite = (ResumableWrite) o;
80+
return Objects.equals(req, resumableWrite.getReq())
81+
&& Objects.equals(res, resumableWrite.getRes());
82+
}
83+
84+
@Override
85+
public int hashCode() {
86+
return Objects.hash(req, res);
87+
}
88+
89+
/**
90+
* Helper function which is more specific than {@link Function#identity()}. Constraining the input
91+
* and output to be exactly {@link BidiResumableWrite}.
92+
*/
93+
static BidiResumableWrite identity(BidiResumableWrite w) {
94+
return w;
95+
}
96+
}

0 commit comments

Comments
 (0)