class GoogleCloudStorageGrpcReadChannel implements SeekableByteChannel {

  @Override
  public int read(ByteBuffer byteBuffer) throws IOException {
    logger.atFine().log(
        "GCS gRPC read request for up to %d bytes at offset %d from object %s",
        byteBuffer.remaining(), position(), objectName);

    if (!isOpen()) {
      throw new ClosedChannelException();
    }

    int bytesRead = 0;

    // The server responds in 2MB chunks, but the client can ask for less than that. We store the
    // remainder in bufferedContent and return pieces of that on the next read call (and flush
    // that buffer if there is a seek).
    if (bufferedContent != null) {
      bytesRead += readBufferedContentInto(byteBuffer);
    }
    if (!byteBuffer.hasRemaining()) {
      return bytesRead;
    }
    if (position == objectSize) {
      return bytesRead > 0 ? bytesRead : -1;
    }
    if (resIterator == null) {
      OptionalLong bytesToRead;
      if (readStrategy == Fadvise.RANDOM) {
        long randomRangeSize =
            Math.max(byteBuffer.remaining(), readOptions.getMinRangeRequestSize());
        bytesToRead = OptionalLong.of(randomRangeSize);
      } else {
        bytesToRead = OptionalLong.empty();
      }
      requestObjectMedia(bytesToRead);
    }
    while (moreServerContent() && byteBuffer.hasRemaining()) {
      GetObjectMediaResponse res = resIterator.next();

      ByteString content = res.getChecksummedData().getContent();
      if (bytesToSkipBeforeReading >= 0 && bytesToSkipBeforeReading < content.size()) {
        content = res.getChecksummedData().getContent().substring((int) bytesToSkipBeforeReading);
        bytesToSkipBeforeReading = 0;
      } else if (bytesToSkipBeforeReading >= content.size()) {
        bytesToSkipBeforeReading -= content.size();
        continue;
      }

      if (readOptions.isGrpcChecksumsEnabled() && res.getChecksummedData().hasCrc32C()) {
        // TODO: Concatenate all these hashes together and compare the result at the end.
        Hasher messageHasher = Hashing.crc32c().newHasher();
        messageHasher.putBytes(res.getChecksummedData().getContent().toByteArray());
        int calculatedChecksum = messageHasher.hash().asInt();
        int expectedChecksum = res.getChecksummedData().getCrc32C().getValue();
        if (calculatedChecksum != expectedChecksum) {
          throw new IOException(
              String.format(
                  "For %s: Message checksum didn't match. Expected %s, got %s.",
                  this, expectedChecksum, calculatedChecksum));
        }
      }

      boolean responseSizeLargerThanRemainingBuffer = content.size() > byteBuffer.remaining();
      int bytesToWrite =
          responseSizeLargerThanRemainingBuffer ? byteBuffer.remaining() : content.size();
      put(content, 0, bytesToWrite, byteBuffer);
      bytesRead += bytesToWrite;
      position += bytesToWrite;

      if (responseSizeLargerThanRemainingBuffer) {
        bufferedContent = content;
        bufferedContentReadOffset = bytesToWrite;
      }
    }
    return bytesRead;
  }

  private  GoogleCloudStorageGrpcReadChannel(
      StorageBlockingStub gcsGrpcBlockingStub,
      String bucketName,
      String objectName,
      long objectGeneration,
      long objectSize,
      GoogleCloudStorageReadOptions readOptions);

}