docs: @Override
    public void writePayload(InputStream message) {
      verifyNotClosed();
      messagesBuffered++;
      currentMessageSeqNo++;
      currentMessageWireSize = 0;
      statsTraceCtx.outboundMessage(currentMessageSeqNo);
      boolean compressed = messageCompression && compressor != Codec.Identity.NONE;
      int written = -1;
      int messageLength = -2;
      try {
        messageLength = getKnownLength(message);
        if (messageLength != 0 && compressed) {
          written = writeCompressed(message, messageLength);
        } else {
          written = writeUncompressed(message, messageLength);
        }
      } catch (IOException e) {
        // This should not be possible, since sink#deliverFrame doesn't throw.
        throw Status.INTERNAL
            .withDescription("Failed to frame message")
            .withCause(e)
            .asRuntimeException();
      } catch (RuntimeException e) {
        throw Status.INTERNAL
            .withDescription("Failed to frame message")
            .withCause(e)
            .asRuntimeException();
      }
  
      if (messageLength != -1 && written != messageLength) {
        String err = String.format("Message length inaccurate %s != %s", written, messageLength);
        throw Status.INTERNAL.withDescription(err).asRuntimeException();
      }
      statsTraceCtx.outboundUncompressedSize(written);
      statsTraceCtx.outboundWireSize(currentMessageWireSize);
      statsTraceCtx.outboundMessageSent(currentMessageSeqNo, currentMessageWireSize, written);
    }