class GoogleCloudStorageReadChannel implements SeekableByteChannel {

  @Override
  public int read(ByteBuffer buffer) throws IOException {
    throwIfNotOpen();

    // Don't try to read if the buffer has no space.
    if (buffer.remaining() == 0) {
      return 0;
    }

    logger.atFine().log(
        "Reading %s bytes at %s position from '%s'",
        buffer.remaining(), currentPosition, resourceId);

    // Do not perform any further reads if we already read everything from this channel.
    if (currentPosition == size) {
      return -1;
    }

    int totalBytesRead = 0;
    int retriesAttempted = 0;

    // We read from a streaming source. We may not get all the bytes we asked for
    // in the first read. Therefore, loop till we either read the required number of
    // bytes or we reach end-of-stream.
    do {
      int remainingBeforeRead = buffer.remaining();
      performLazySeek(remainingBeforeRead);
      checkState(
          contentChannelPosition == currentPosition,
          "contentChannelPosition (%s) should be equal to currentPosition (%s) after lazy seek",
          contentChannelPosition, currentPosition);

      try {
        int numBytesRead = contentChannel.read(buffer);
        checkIOPrecondition(numBytesRead != 0, "Read 0 bytes without blocking");
        if (numBytesRead < 0) {
          // Because we don't know decompressed object size for gzip-encoded objects,
          // assume that this is an object end.
          if (gzipEncoded) {
            size = currentPosition;
            contentChannelEnd = currentPosition;
          }
          // Check that we didn't get a premature End of Stream signal by checking the number of
          // bytes read against the stream size. Unfortunately we don't have information about the
          // actual size of the data stream when stream compression is used, so we can only ignore
          // this case here.
          checkIOPrecondition(
              currentPosition == contentChannelEnd || currentPosition == size,
              String.format(
                  "Received end of stream result before all the file data has been received; "
                      + "totalBytesRead: %d, currentPosition: %d,"
                      + " contentChannelEnd %d, size: %d, object: '%s'",
                  totalBytesRead, currentPosition, contentChannelEnd, size, resourceId));

          // If we have reached an end of a contentChannel but not an end of an object
          // then close contentChannel and continue reading an object if necessary.
          if (contentChannelEnd != size && currentPosition == contentChannelEnd) {
            closeContentChannel();
          } else {
            break;
          }
        }

        if (numBytesRead > 0) {
          totalBytesRead += numBytesRead;
          currentPosition += numBytesRead;
          contentChannelPosition += numBytesRead;
          checkState(
              contentChannelPosition == currentPosition,
              "contentChannelPosition (%s) should be equal to currentPosition (%s)"
                  + " after successful read",
              contentChannelPosition, currentPosition);
        }

        if (retriesAttempted != 0) {
          logger.atInfo().log(
              "Success after %s retries on reading '%s'", retriesAttempted, resourceId);
        }
        // The count of retriesAttempted is per low-level contentChannel.read call;
        // each time we make progress we reset the retry counter.
        retriesAttempted = 0;
      } catch (IOException ioe) {
        logger.atFine().log(
            "Closing contentChannel after %s exception for '%s'.", ioe.getMessage(), resourceId);
        closeContentChannel();

        if (buffer.remaining() != remainingBeforeRead) {
          int partialRead = remainingBeforeRead - buffer.remaining();
          logger.atInfo().log(
              "Despite exception, had partial read of %s bytes from '%s'; resetting retry count.",
              partialRead, resourceId);
          retriesAttempted = 0;
          totalBytesRead += partialRead;
          currentPosition += partialRead;
        }

        // TODO(user): Refactor any reusable logic for retries into a separate RetryHelper class.
        if (retriesAttempted == maxRetries) {
          logger.atSevere().log(
              "Throwing exception after reaching max read retries (%s) for '%s'.",
              maxRetries, resourceId);
          throw ioe;
        }

        if (retriesAttempted == 0) {
          // If this is the first of a series of retries, we also want to reset the readBackOff
          // to have fresh initial values.
          readBackOff.get().reset();
        }

        ++retriesAttempted;
        logger.atWarning().withCause(ioe).log(
            "Failed read retry #%s/%s for '%s'. Sleeping...",
            retriesAttempted, maxRetries, resourceId);
        try {
          boolean backOffSuccessful = BackOffUtils.next(sleeper, readBackOff.get());
          if (!backOffSuccessful) {
            logger.atSevere().log(
                "BackOff returned false; maximum total elapsed time exhausted."
                    + " Giving up after %s/%s retries for '%s'",
                retriesAttempted, maxRetries, resourceId);
            throw ioe;
          }
        } catch (InterruptedException ie) {
          Thread.currentThread().interrupt();
          logger.atSevere().log(
              "Interrupted while sleeping before retry. Giving up after %s/%s retries for '%s'",
              retriesAttempted, maxRetries, resourceId);
          ioe.addSuppressed(ie);
          throw ioe;
        }

        logger.atInfo().log(
            "Done sleeping before retry #%s/%s for '%s'", retriesAttempted, maxRetries, resourceId);
      } catch (RuntimeException r) {
        closeContentChannel();
        throw r;
      }
    } while (buffer.remaining() > 0 && currentPosition < size);

    // If this method was called when the stream was already at EOF
    // (indicated by totalBytesRead == 0) then return EOF else,
    // return the number of bytes read.
    boolean isEndOfStream = (totalBytesRead == 0);
    if (isEndOfStream) {
      // Check that we didn't get a premature End of Stream signal by checking the number of bytes
      // read against the stream size. Unfortunately we don't have information about the actual size
      // of the data stream when stream compression is used, so we can only ignore this case here.
      checkIOPrecondition(
          currentPosition == size,
          String.format(
              "Failed to read any data before all the file data has been received;"
                  + " currentPosition: %d, size: %d, object '%s'",
              currentPosition, size, resourceId));
      return -1;
    }
    return totalBytesRead;
  }

  public  GoogleCloudStorageReadChannel(
      Storage gcs,
      String bucketName,
      String objectName,
      ApiErrorExtractor errorExtractor,
      ClientRequestHelper<StorageObject> requestHelper);
  public  GoogleCloudStorageReadChannel(
      Storage gcs,
      String bucketName,
      String objectName,
      ApiErrorExtractor errorExtractor,
      ClientRequestHelper<StorageObject> requestHelper,
      @Nonnull GoogleCloudStorageReadOptions readOptions);
  public  GoogleCloudStorageReadChannel(
      Storage gcs,
      StorageResourceId resourceId,
      ApiErrorExtractor errorExtractor,
      ClientRequestHelper<StorageObject> requestHelper,
      @Nonnull GoogleCloudStorageReadOptions readOptions);
  protected  GoogleCloudStorageReadChannel(
      String bucketName, String objectName, @Nonnull GoogleCloudStorageReadOptions readOptions);

}