com.google.protobuf.ByteString#writeTo ( )源码实例Demo

public static ListenableFuture<Digest> putBlobFuture(
    Instance instance,
    Digest digest,
    ByteString data,
    long writeDeadlineAfter,
    TimeUnit writeDeadlineAfterUnits,
    RequestMetadata requestMetadata)
    throws ExcessiveWriteSizeException {
  if (digest.getSizeBytes() != data.size()) {
    return immediateFailedFuture(
        invalidDigestSize(digest.getSizeBytes(), data.size()).asRuntimeException());
  }
  Write write = instance.getBlobWrite(digest, UUID.randomUUID(), requestMetadata);
  // indicate that we know this write is novel
  write.reset();
  SettableFuture<Digest> future = SettableFuture.create();
  write.addListener(() -> future.set(digest), directExecutor());
  try (OutputStream out =
      write.getOutput(writeDeadlineAfter, writeDeadlineAfterUnits, () -> {})) {
    data.writeTo(out);
  } catch (Exception e) {
    future.setException(e);
  }
  return future;
}
 

static ByteString getBlobIntoFile(
    String type, String instanceName, Digest digest, ByteStreamStub bsStub, Path root)
    throws IOException, InterruptedException {
  Path file = root.resolve(digest.getHash());
  if (Files.exists(file) && Files.size(file) == digest.getSizeBytes()) {
    try (InputStream in = Files.newInputStream(file)) {
      return ByteString.readFrom(in);
    }
  }
  System.out.println("Getting " + type + " " + digest.getHash() + "/" + digest.getSizeBytes());
  ByteString content = getBlob(instanceName, digest, bsStub);
  try (OutputStream out = Files.newOutputStream(file)) {
    content.writeTo(out);
  }
  return content;
}
 

@Test
public void writeAddsEntry() throws IOException {
  ByteString content = ByteString.copyFromUtf8("Hello, World");
  Digest digest = DIGEST_UTIL.compute(content);

  AtomicBoolean notified = new AtomicBoolean(false);
  Write write = getWrite(digest);
  write.addListener(() -> notified.set(true), directExecutor());
  try (OutputStream out = write.getOutput(1, SECONDS, () -> {})) {
    content.writeTo(out);
  }
  assertThat(notified.get()).isTrue();
  String key = fileCache.getKey(digest, false);
  assertThat(storage.get(key)).isNotNull();
  try (InputStream in = Files.newInputStream(fileCache.getPath(key))) {
    assertThat(ByteString.readFrom(in)).isEqualTo(content);
  }
}
 

@Test
public void asyncWriteCompletionDischargesWriteSize() throws IOException {
  ByteString content = ByteString.copyFromUtf8("Hello, World");
  Digest digest = DIGEST_UTIL.compute(content);

  Write completingWrite = getWrite(digest);
  Write incompleteWrite = getWrite(digest);
  AtomicBoolean notified = new AtomicBoolean(false);
  // both should be size committed
  incompleteWrite.addListener(() -> notified.set(true), directExecutor());
  OutputStream incompleteOut = incompleteWrite.getOutput(1, SECONDS, () -> {});
  try (OutputStream out = completingWrite.getOutput(1, SECONDS, () -> {})) {
    assertThat(fileCache.size()).isEqualTo(digest.getSizeBytes() * 2);
    content.writeTo(out);
  }
  assertThat(notified.get()).isTrue();
  assertThat(fileCache.size()).isEqualTo(digest.getSizeBytes());
  assertThat(incompleteWrite.getCommittedSize()).isEqualTo(digest.getSizeBytes());
  assertThat(incompleteWrite.isComplete()).isTrue();
  incompleteOut.close(); // redundant
}
 

@Test
public void resetIsRespectedOnSubsequentWrite() throws IOException {
  String resourceName = "reset-resource";
  StubWriteOutputStream write =
      new StubWriteOutputStream(
          Suppliers.ofInstance(ByteStreamGrpc.newBlockingStub(channel)),
          Suppliers.ofInstance(ByteStreamGrpc.newStub(channel)),
          resourceName,
          /* expectedSize=*/ StubWriteOutputStream.UNLIMITED_EXPECTED_SIZE,
          /* autoflush=*/ true);
  ByteString content = ByteString.copyFromUtf8("Hello, World");
  try (OutputStream out = write.getOutput(1, SECONDS, () -> {})) {
    content.writeTo(out);
    write.reset();
    content.writeTo(out);
  }
  verify(serviceImpl, times(1)).write(any(StreamObserver.class));
  ArgumentCaptor<WriteRequest> writeRequestCaptor = ArgumentCaptor.forClass(WriteRequest.class);
  verify(writeObserver, times(3)).onNext(writeRequestCaptor.capture());
  List<WriteRequest> requests = writeRequestCaptor.getAllValues();
  assertThat(requests.get(0).getWriteOffset()).isEqualTo(requests.get(1).getWriteOffset());
  assertThat(requests.get(2).getFinishWrite()).isTrue();
}
 

private void writeData(ByteString data) {
  try {
    data.writeTo(getOutput());
    requestNextIfReady();
  } catch (IOException e) {
    if (!committed) {
      logger.log(Level.SEVERE, format("error writing data for %s", name), e);
      responseObserver.onError(Status.fromThrowable(e).asException());
    }
    // shouldn't we be erroring the stream at this point if !committed?
  }
}
 

public HashCode computeHash(ByteString blob) {
  Hasher hasher = hashFn.getHash().newHasher();
  try {
    blob.writeTo(Funnels.asOutputStream(hasher));
  } catch (IOException e) {
    /* impossible, due to Funnels.asOutputStream behavior */
  }
  return hasher.hash();
}
 

@Override
public void onNext(ReadResponse response) {
  ByteString data = response.getData();
  try {
    data.writeTo(sink);
  } catch (IOException e) {
    throw new RuntimeException(e);
  }
  sizes.add(data.size());
}
 

@Override
public void encode(ByteString value, OutputStream outStream, Context context)
    throws IOException, CoderException {
  if (value == null) {
    throw new CoderException("cannot encode a null ByteString");
  }

  if (!context.isWholeStream) {
    // ByteString is not delimited, so write its size before its contents.
    VarInt.encode(value.size(), outStream);
  }
  value.writeTo(outStream);
}
 

public void testByteStringToOutputStream() throws Exception {
  byte[] randomBytes = readStream(getTestData("randombytes"));
  ByteString bs = ByteString.copyFrom(randomBytes);
  ByteArrayOutputStream out = new ByteArrayOutputStream();
  bs.writeTo(out);
  checkBytes(randomBytes, out.toByteArray());
}
 

private void writeByteArray(ByteString bb)
  throws java.io.IOException
{
  if (bb == null)
  {
    writeNull();
    return;
  }
  byte[] bytes = bb.toByteArray();
  writeInt(bb.size());
  bb.writeTo(sock.getOutputStream());
}
 

private void writeXmlFrom(XmlElement element) throws IOException {
  ANGLE_OPEN.writeTo(out);
  if (!element.getNamespaceUriBytes().isEmpty()) {
    findNamespacePrefix(element.getNamespaceUriBytes()).writeTo(out);
    COLON.writeTo(out);
  }
  final ByteString name = element.getNameBytes();
  name.writeTo(out);
  final Map<ByteString, ByteString> namespaces = new LinkedHashMap<>();
  for (XmlNamespace namespace : element.getNamespaceDeclarationList()) {
    final ByteString prefix = namespace.getPrefixBytes();
    SPACE.writeTo(out);
    XMLNS.writeTo(out);
    prefix.writeTo(out);
    EQUALS.writeTo(out);
    quote(namespace.getUriBytes());
    namespaces.put(namespace.getUriBytes(), prefix);
  }
  namespaceStack.push(namespaces);
  for (XmlAttribute attribute : element.getAttributeList()) {
    SPACE.writeTo(out);
    if (!attribute.getNamespaceUriBytes().isEmpty()) {
      findNamespacePrefix(attribute.getNamespaceUriBytes()).writeTo(out);
      COLON.writeTo(out);
    }
    attribute.getNameBytes().writeTo(out);
    EQUALS.writeTo(out);
    quote(attribute.getValueBytes());
  }
  if (element.getChildList().isEmpty()) {
    FORWARD_SLASH.writeTo(out);
    ANGLE_CLOSE.writeTo(out);
  } else {
    ANGLE_CLOSE.writeTo(out);
    for (XmlNode child : element.getChildList()) {
      writeXmlFrom(child);
    }
    ANGLE_OPEN.writeTo(out);
    FORWARD_SLASH.writeTo(out);
    if (!element.getNamespaceUriBytes().isEmpty()) {
      findNamespacePrefix(element.getNamespaceUriBytes()).writeTo(out);
      COLON.writeTo(out);
    }
    name.writeTo(out);
    ANGLE_CLOSE.writeTo(out);
  }
  namespaceStack.pop();
}