fix: update grpc bidi resumable uploads to validate ack'd object size

Follow up to #2527

Part 1, add tests and simulation server

Author: BenWhitehead

google-cloud-storage/src/main/java/com/google/cloud/storage/GapicBidiUnbufferedWritableByteChannel.java

@@ -66,17 +66,17 @@ final class GapicBidiUnbufferedWritableByteChannel implements UnbufferedWritable
       ResultRetryAlgorithm<?> alg,
       SettableApiFuture<BidiWriteObjectResponse> resultFuture,
       ChunkSegmenter chunkSegmenter,
-      BidiResumableWrite requestFactory,
+      BidiWriteCtx<BidiResumableWrite> writeCtx,
       Supplier<GrpcCallContext> baseContextSupplier) {
     this.write = write;
     this.deps = deps;
     this.alg = alg;
     this.baseContextSupplier = baseContextSupplier;
-    this.bucketName = requestFactory.bucketName();
+    this.bucketName = writeCtx.getRequestFactory().bucketName();
     this.resultFuture = resultFuture;
     this.chunkSegmenter = chunkSegmenter;
 
-    this.writeCtx = new BidiWriteCtx<>(requestFactory);
+    this.writeCtx = writeCtx;
     this.responseObserver = new BidiObserver();
   }
 

google-cloud-storage/src/main/java/com/google/cloud/storage/GapicBidiWritableByteChannelSessionBuilder.java

@@ -155,7 +155,7 @@ BufferedWritableByteChannelSession<BidiWriteObjectResponse> build() {
                             resultFuture,
                             new ChunkSegmenter(
                                 boundHasher, boundStrategy, Values.MAX_WRITE_CHUNK_BYTES_VALUE),
-                            start,
+                            new BidiWriteCtx<>(start),
                             Retrying::newCallContext))
                 .andThen(c -> new DefaultBufferedWritableByteChannel(bufferHandle, c))
                 .andThen(StorageByteChannels.writable()::createSynchronized));