@Test
public void batchUpdateBlobs() {
  DigestUtil digestUtil = new DigestUtil(HashFunction.SHA256);
  ByteString content = ByteString.copyFromUtf8("Hello, World!");
  Digest digest = digestUtil.compute(content);
  BatchUpdateBlobsRequest request =
      BatchUpdateBlobsRequest.newBuilder()
          .setInstanceName(INSTANCE_NAME)
          .addRequests(Request.newBuilder().setDigest(digest).setData(content).build())
          .build();
  ContentAddressableStorageGrpc.ContentAddressableStorageBlockingStub stub =
      ContentAddressableStorageGrpc.newBlockingStub(inProcessChannel);

  BatchUpdateBlobsResponse response = stub.batchUpdateBlobs(request);

  Response expected =
      Response.newBuilder()
          .setDigest(digest)
          .setStatus(com.google.rpc.Status.newBuilder().setCode(Code.OK.getNumber()).build())
          .build();
  assertThat(response.getResponsesList())
      .containsExactlyElementsIn(Collections.singleton(expected));
}
 

@Override
public void batchUpdateBlobs(
    BatchUpdateBlobsRequest request, StreamObserver<BatchUpdateBlobsResponse> responseObserver) {
  BatchUpdateBlobsResponse.Builder batchResponse = BatchUpdateBlobsResponse.newBuilder();
  for (BatchUpdateBlobsRequest.Request r : request.getRequestsList()) {
    BatchUpdateBlobsResponse.Response.Builder resp = batchResponse.addResponsesBuilder();
    try {
      Digest digest = cache.getDigestUtil().compute(r.getData().toByteArray());
      getFromFuture(cache.uploadBlob(digest, r.getData()));
      if (!r.getDigest().equals(digest)) {
        String err =
            "Upload digest " + r.getDigest() + " did not match data digest: " + digest;
        resp.setStatus(StatusUtils.invalidArgumentStatus("digest", err));
        continue;
      }
      resp.getStatusBuilder().setCode(Code.OK.getNumber());
    } catch (Exception e) {
      resp.setStatus(StatusUtils.internalErrorStatus(e));
    }
  }
  responseObserver.onNext(batchResponse.build());
  responseObserver.onCompleted();
}
 

@Override
public Iterable<Digest> putAllBlobs(Iterable<ByteString> blobs, RequestMetadata requestMetadata) {
  long totalSize = 0;
  ImmutableList.Builder<Request> requests = ImmutableList.builder();
  for (ByteString blob : blobs) {
    checkState(totalSize + blob.size() <= maxBatchUpdateBlobsSize);
    requests.add(Request.newBuilder().setDigest(digestUtil.compute(blob)).setData(blob).build());
    totalSize += blob.size();
  }
  BatchUpdateBlobsRequest batchRequest =
      BatchUpdateBlobsRequest.newBuilder()
          .setInstanceName(getName())
          .addAllRequests(requests.build())
          .build();
  BatchUpdateBlobsResponse batchResponse =
      deadlined(casBlockingStub)
          .withInterceptors(attachMetadataInterceptor(requestMetadata))
          .batchUpdateBlobs(batchRequest);
  PutAllBlobsException exception = null;
  for (BatchUpdateBlobsResponse.Response response : batchResponse.getResponsesList()) {
    com.google.rpc.Status status = response.getStatus();
    if (Code.forNumber(status.getCode()) != Code.OK) {
      if (exception == null) {
        exception = new PutAllBlobsException();
      }
      exception.addFailedResponse(response);
    }
  }
  if (exception != null) {
    throw exception;
  }
  return Iterables.transform(
      batchResponse.getResponsesList(), (response) -> response.getDigest());
}
 

@Test
public void putAllBlobsUploadsBlobs() throws Exception {
  String instanceName = "putAllBlobs-test";
  serviceRegistry.addService(
      new ContentAddressableStorageImplBase() {
        @Override
        public void batchUpdateBlobs(
            BatchUpdateBlobsRequest batchRequest,
            StreamObserver<BatchUpdateBlobsResponse> responseObserver) {
          checkState(batchRequest.getInstanceName().equals(instanceName));
          responseObserver.onNext(
              BatchUpdateBlobsResponse.newBuilder()
                  .addAllResponses(
                      Iterables.transform(
                          batchRequest.getRequestsList(),
                          request ->
                              Response.newBuilder().setDigest(request.getDigest()).build()))
                  .build());
          responseObserver.onCompleted();
        }
      });
  Instance instance = newStubInstance("putAllBlobs-test");
  ByteString first = ByteString.copyFromUtf8("first");
  ByteString last = ByteString.copyFromUtf8("last");
  ImmutableList<ByteString> blobs = ImmutableList.of(first, last);
  ImmutableList<Digest> digests =
      ImmutableList.of(DIGEST_UTIL.compute(first), DIGEST_UTIL.compute(last));
  assertThat(instance.putAllBlobs(blobs, RequestMetadata.getDefaultInstance()))
      .containsAllIn(digests);
}
 

@Override
public void batchUpdateBlobs(
    BatchUpdateBlobsRequest request, StreamObserver<BatchUpdateBlobsResponse> responseObserver) {
  try {
    ImmutableList<UploadResult> uploadResults =
        storage.batchUpdateBlobs(
            request.getRequestsList().stream()
                .map(
                    blobRequest ->
                        UploadDataSupplier.of(
                            blobRequest.toString(),
                            new GrpcDigest(blobRequest.getDigest()),
                            () -> new ByteArrayInputStream(blobRequest.getData().toByteArray())))
                .collect(ImmutableList.toImmutableList()));

    BatchUpdateBlobsResponse.Builder responseBuilder = BatchUpdateBlobsResponse.newBuilder();
    for (UploadResult uploadResult : uploadResults) {
      Builder statusBuilder = com.google.rpc.Status.newBuilder();
      statusBuilder.setCode(uploadResult.status);
      if (uploadResult.status != 0) {
        statusBuilder.setMessage(uploadResult.message);
      }
      responseBuilder.addResponses(
          Response.newBuilder()
              .setDigest(GrpcProtocol.get(uploadResult.digest))
              .setStatus(statusBuilder.build())
              .build());
    }
    responseObserver.onNext(responseBuilder.build());
    responseObserver.onCompleted();
  } catch (Exception e) {
    // idk how this should be done
    e.printStackTrace();
    responseObserver.onError(new StatusRuntimeException(Status.fromThrowable(e)));
  }
}