Skip to content

Commit acff0cd

Browse files
BenWhiteheadBenWhitehead
committed
chore: remove old appendble upload implementation
1 parent fb34cb2 commit acff0cd

8 files changed

Lines changed: 67 additions & 1122 deletions

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

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -17,8 +17,8 @@
1717
package com.google.cloud.storage;
1818

1919
import com.google.cloud.BaseServiceException;
20-
import com.google.cloud.storage.BlobAppendableUploadImpl.AppendableUnbufferedWritableByteChannel;
2120
import com.google.cloud.storage.ChunkSegmenter.ChunkSegment;
21+
import com.google.cloud.storage.UnbufferedWritableByteChannelSession.UnbufferedWritableByteChannel;
2222
import java.io.IOException;
2323
import java.io.InterruptedIOException;
2424
import java.nio.ByteBuffer;
@@ -27,8 +27,7 @@
2727
import java.util.concurrent.TimeUnit;
2828
import java.util.concurrent.TimeoutException;
2929

30-
final class BidiAppendableUnbufferedWritableByteChannel
31-
implements AppendableUnbufferedWritableByteChannel {
30+
final class BidiAppendableUnbufferedWritableByteChannel implements UnbufferedWritableByteChannel {
3231

3332
private final BidiUploadStreamingStream stream;
3433
private final ChunkSegmenter chunkSegmenter;

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

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -111,7 +111,8 @@ public WritableByteChannelSession<?, BlobInfo> writeSession(
111111
GrpcStorageImpl grpc = (GrpcStorageImpl) s;
112112
GrpcCallContext grpcCallContext =
113113
opts.grpcMetadataMapper().apply(GrpcCallContext.createDefault());
114-
BidiWriteObjectRequest req = grpc.getBidiWriteObjectRequest(info, opts);
114+
BidiWriteObjectRequest req =
115+
grpc.getBidiWriteObjectRequest(info, opts, false);
115116

116117
ApiFuture<BidiResumableWrite> startResumableWrite =
117118
grpc.startResumableWrite(grpcCallContext, req, opts);

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

Lines changed: 0 additions & 70 deletions
Original file line numberDiff line numberDiff line change
@@ -94,73 +94,3 @@ static BidiResumableWrite identity(BidiResumableWrite w) {
9494
return w;
9595
}
9696
}
97-
98-
final class BidiAppendableWrite implements BidiWriteObjectRequestBuilderFactory {
99-
100-
private final BidiWriteObjectRequest req;
101-
102-
public BidiAppendableWrite(BidiWriteObjectRequest req) {
103-
this(req, false);
104-
}
105-
106-
public BidiAppendableWrite(BidiWriteObjectRequest req, boolean takeOver) {
107-
if (takeOver) {
108-
this.req = req;
109-
} else {
110-
req =
111-
req.toBuilder()
112-
.setWriteObjectSpec(req.getWriteObjectSpec().toBuilder().setAppendable(true).build())
113-
.build();
114-
this.req = req;
115-
}
116-
}
117-
118-
public BidiWriteObjectRequest getReq() {
119-
return req;
120-
}
121-
122-
@Override
123-
public BidiWriteObjectRequest.Builder newBuilder() {
124-
return req.toBuilder();
125-
}
126-
127-
@Override
128-
public @Nullable String bucketName() {
129-
if (req.hasWriteObjectSpec() && req.getWriteObjectSpec().hasResource()) {
130-
return req.getWriteObjectSpec().getResource().getBucket();
131-
} else if (req.hasAppendObjectSpec()) {
132-
return req.getAppendObjectSpec().getBucket();
133-
}
134-
return null;
135-
}
136-
137-
@Override
138-
public String toString() {
139-
return "BidiAppendableWrite{" + "req=" + fmtProto(req) + '}';
140-
}
141-
142-
@Override
143-
public boolean equals(Object o) {
144-
if (this == o) {
145-
return true;
146-
}
147-
if (!(o instanceof BidiAppendableWrite)) {
148-
return false;
149-
}
150-
BidiAppendableWrite BidiAppendableWrite = (BidiAppendableWrite) o;
151-
return Objects.equals(req, BidiAppendableWrite.getReq());
152-
}
153-
154-
@Override
155-
public int hashCode() {
156-
return Objects.hash(req);
157-
}
158-
159-
/**
160-
* Helper function which is more specific than {@link Function#identity()}. Constraining the input
161-
* and output to be exactly {@link BidiAppendableWrite}.
162-
*/
163-
static BidiAppendableWrite identity(BidiAppendableWrite w) {
164-
return w;
165-
}
166-
}

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

Lines changed: 47 additions & 118 deletions
Original file line numberDiff line numberDiff line change
@@ -24,9 +24,6 @@
2424
import com.google.api.core.BetaApi;
2525
import com.google.api.core.InternalApi;
2626
import com.google.api.core.SettableApiFuture;
27-
import com.google.api.gax.retrying.BasicResultRetryAlgorithm;
28-
import com.google.api.gax.rpc.AbortedException;
29-
import com.google.api.gax.rpc.ApiException;
3027
import com.google.cloud.storage.BidiUploadState.AppendableUploadState;
3128
import com.google.cloud.storage.BidiUploadState.TakeoverAppendableUploadState;
3229
import com.google.cloud.storage.BlobAppendableUpload.AppendableUploadWriteableByteChannel;
@@ -36,9 +33,7 @@
3633
import com.google.cloud.storage.UnifiedOpts.ObjectTargetOpt;
3734
import com.google.cloud.storage.UnifiedOpts.Opts;
3835
import com.google.common.base.MoreObjects;
39-
import com.google.storage.v2.BidiWriteObjectRequest;
4036
import com.google.storage.v2.BidiWriteObjectResponse;
41-
import com.google.storage.v2.Object;
4237
import com.google.storage.v2.ServiceConstants.Values;
4338
import java.util.function.BiFunction;
4439
import javax.annotation.concurrent.Immutable;
@@ -60,20 +55,17 @@ public final class BlobAppendableUploadConfig {
6055
new BlobAppendableUploadConfig(
6156
FlushPolicy.minFlushSize(_256KiB),
6257
Hasher.enabled(),
63-
CloseAction.CLOSE_WITHOUT_FINALIZING,
64-
false);
58+
CloseAction.CLOSE_WITHOUT_FINALIZING);
6559

6660
private final FlushPolicy flushPolicy;
6761
private final Hasher hasher;
6862
private final CloseAction closeAction;
69-
private final boolean newImpl;
7063

7164
private BlobAppendableUploadConfig(
72-
FlushPolicy flushPolicy, Hasher hasher, CloseAction closeAction, boolean newImpl) {
65+
FlushPolicy flushPolicy, Hasher hasher, CloseAction closeAction) {
7366
this.flushPolicy = flushPolicy;
7467
this.hasher = hasher;
7568
this.closeAction = closeAction;
76-
this.newImpl = newImpl;
7769
}
7870

7971
/**
@@ -104,7 +96,7 @@ public BlobAppendableUploadConfig withFlushPolicy(FlushPolicy flushPolicy) {
10496
if (this.flushPolicy.equals(flushPolicy)) {
10597
return this;
10698
}
107-
return new BlobAppendableUploadConfig(flushPolicy, hasher, closeAction, newImpl);
99+
return new BlobAppendableUploadConfig(flushPolicy, hasher, closeAction);
108100
}
109101

110102
/**
@@ -134,7 +126,7 @@ public BlobAppendableUploadConfig withCloseAction(CloseAction closeAction) {
134126
if (this.closeAction == closeAction) {
135127
return this;
136128
}
137-
return new BlobAppendableUploadConfig(flushPolicy, hasher, closeAction, newImpl);
129+
return new BlobAppendableUploadConfig(flushPolicy, hasher, closeAction);
138130
}
139131

140132
/**
@@ -166,7 +158,7 @@ BlobAppendableUploadConfig withCrc32cValidationEnabled(boolean enabled) {
166158
return this;
167159
}
168160
return new BlobAppendableUploadConfig(
169-
flushPolicy, enabled ? Hasher.enabled() : Hasher.noop(), closeAction, newImpl);
161+
flushPolicy, enabled ? Hasher.enabled() : Hasher.noop(), closeAction);
170162
}
171163

172164
/** Never to be made public until {@link Hasher} is public */
@@ -175,10 +167,6 @@ Hasher getHasher() {
175167
return hasher;
176168
}
177169

178-
BlobAppendableUploadConfig useNewImpl() {
179-
return new BlobAppendableUploadConfig(flushPolicy, hasher, closeAction, true);
180-
}
181-
182170
@Override
183171
public String toString() {
184172
return MoreObjects.toStringHelper(this)
@@ -239,109 +227,50 @@ public enum CloseAction {
239227
}
240228

241229
BlobAppendableUpload create(GrpcStorageImpl storage, BlobInfo info, Opts<ObjectTargetOpt> opts) {
242-
if (newImpl) {
243-
// TODO: make configurable
244-
int maxRedirectsAllowed = 3;
245-
246-
long maxPendingBytes = this.getFlushPolicy().getMaxPendingBytes();
247-
AppendableUploadState state = storage.getAppendableState(info, opts, maxPendingBytes);
248-
WritableByteChannelSession<
249-
AppendableObjectBufferedWritableByteChannel, BidiWriteObjectResponse>
250-
build =
251-
new AppendableSession(
252-
ApiFutures.immediateFuture(state),
253-
(start, resultFuture) -> {
254-
BidiUploadStreamingStream stream =
255-
new BidiUploadStreamingStream(
256-
start,
257-
storage.storageDataClient.executor,
258-
storage.storageClient.bidiWriteObjectCallable(),
259-
maxRedirectsAllowed,
260-
storage.storageDataClient.retryContextProvider.create());
261-
ChunkSegmenter chunkSegmenter =
262-
new ChunkSegmenter(
263-
Hasher.enabled(),
264-
ByteStringStrategy.copy(),
265-
Math.min(
266-
Values.MAX_WRITE_CHUNK_BYTES_VALUE,
267-
Math.toIntExact(maxPendingBytes)),
268-
/* blockSize= */ 1);
269-
BidiAppendableUnbufferedWritableByteChannel c;
270-
if (state instanceof TakeoverAppendableUploadState) {
271-
// start the takeover reconciliation
272-
stream.awaitTakeoverStateReconciliation();
273-
c =
274-
new BidiAppendableUnbufferedWritableByteChannel(
275-
stream, chunkSegmenter, state.getConfirmedBytes());
276-
} else {
277-
c =
278-
new BidiAppendableUnbufferedWritableByteChannel(
279-
stream, chunkSegmenter, 0);
280-
}
281-
return new AppendableObjectBufferedWritableByteChannel(
282-
flushPolicy.createBufferedChannel(c),
283-
c,
284-
this.closeAction == CloseAction.FINALIZE_WHEN_CLOSING,
285-
newImpl);
286-
},
287-
state.getResultFuture());
230+
// TODO: make configurable
231+
int maxRedirectsAllowed = 3;
288232

289-
return new BlobAppendableUploadImpl(
290-
new DefaultBlobWriteSessionConfig.DecoratedWritableByteChannelSession<>(
291-
build, BidiBlobWriteSessionConfig.Factory.WRITE_OBJECT_RESPONSE_BLOB_INFO_DECODER));
292-
} else {
293-
boolean takeOver = info.getGeneration() != null;
294-
BidiWriteObjectRequest req =
295-
takeOver
296-
? storage.getBidiWriteObjectRequestForTakeover(info, opts)
297-
: storage.getBidiWriteObjectRequest(info, opts);
233+
long maxPendingBytes = this.getFlushPolicy().getMaxPendingBytes();
234+
AppendableUploadState state = storage.getAppendableState(info, opts, maxPendingBytes);
235+
WritableByteChannelSession<AppendableObjectBufferedWritableByteChannel, BidiWriteObjectResponse>
236+
build =
237+
new AppendableSession(
238+
ApiFutures.immediateFuture(state),
239+
(start, resultFuture) -> {
240+
BidiUploadStreamingStream stream =
241+
new BidiUploadStreamingStream(
242+
start,
243+
storage.storageDataClient.executor,
244+
storage.storageClient.bidiWriteObjectCallable(),
245+
maxRedirectsAllowed,
246+
storage.storageDataClient.retryContextProvider.create());
247+
ChunkSegmenter chunkSegmenter =
248+
new ChunkSegmenter(
249+
Hasher.enabled(),
250+
ByteStringStrategy.copy(),
251+
Math.min(
252+
Values.MAX_WRITE_CHUNK_BYTES_VALUE, Math.toIntExact(maxPendingBytes)),
253+
/* blockSize= */ 1);
254+
BidiAppendableUnbufferedWritableByteChannel c;
255+
if (state instanceof TakeoverAppendableUploadState) {
256+
// start the takeover reconciliation
257+
stream.awaitTakeoverStateReconciliation();
258+
c =
259+
new BidiAppendableUnbufferedWritableByteChannel(
260+
stream, chunkSegmenter, state.getConfirmedBytes());
261+
} else {
262+
c = new BidiAppendableUnbufferedWritableByteChannel(stream, chunkSegmenter, 0);
263+
}
264+
return new AppendableObjectBufferedWritableByteChannel(
265+
flushPolicy.createBufferedChannel(c),
266+
c,
267+
this.closeAction == CloseAction.FINALIZE_WHEN_CLOSING);
268+
},
269+
state.getResultFuture());
298270

299-
BidiAppendableWrite baw = new BidiAppendableWrite(req, takeOver);
300-
301-
WritableByteChannelSession<
302-
AppendableObjectBufferedWritableByteChannel, BidiWriteObjectResponse>
303-
build =
304-
ResumableMedia.gapic()
305-
.write()
306-
.bidiByteChannel(storage.storageClient.bidiWriteObjectCallable())
307-
.setHasher(this.getHasher())
308-
.setByteStringStrategy(ByteStringStrategy.copy())
309-
.appendable()
310-
.withRetryConfig(
311-
storage.retrier.withAlg(
312-
new BasicResultRetryAlgorithm<Object>() {
313-
@Override
314-
public boolean shouldRetry(
315-
Throwable previousThrowable, Object previousResponse) {
316-
// TODO: remove this later once the redirects are not handled by the
317-
// retry loop
318-
ApiException apiEx = null;
319-
if (previousThrowable instanceof StorageException) {
320-
StorageException se = (StorageException) previousThrowable;
321-
Throwable cause = se.getCause();
322-
if (cause instanceof ApiException) {
323-
apiEx = (ApiException) cause;
324-
}
325-
}
326-
if (apiEx instanceof AbortedException) {
327-
return true;
328-
}
329-
return storage
330-
.retryAlgorithmManager
331-
.idempotent()
332-
.shouldRetry(previousThrowable, null);
333-
}
334-
}))
335-
.buffered(this.getFlushPolicy())
336-
.setStartAsync(ApiFutures.immediateFuture(baw))
337-
.setGetCallable(storage.storageClient.getObjectCallable())
338-
.setFinalizeOnClose(this.closeAction == CloseAction.FINALIZE_WHEN_CLOSING)
339-
.build();
340-
341-
return new BlobAppendableUploadImpl(
342-
new DefaultBlobWriteSessionConfig.DecoratedWritableByteChannelSession<>(
343-
build, BidiBlobWriteSessionConfig.Factory.WRITE_OBJECT_RESPONSE_BLOB_INFO_DECODER));
344-
}
271+
return new BlobAppendableUploadImpl(
272+
new DefaultBlobWriteSessionConfig.DecoratedWritableByteChannelSession<>(
273+
build, BidiBlobWriteSessionConfig.Factory.WRITE_OBJECT_RESPONSE_BLOB_INFO_DECODER));
345274
}
346275

347276
private static final class AppendableSession

0 commit comments

Comments
 (0)