@Override
  public Future<Long> readInto(AsyncWritableChannel asyncWritableChannel, Callback<Long> callback) {
    Future<Long> future;
    if (!channelOpen.get()) {
      ClosedChannelException closedChannelException = new ClosedChannelException();
      FutureResult<Long> futureResult = new FutureResult<Long>();
      futureResult.done(0L, closedChannelException);
      future = futureResult;
      if (callback != null) {
        callback.onCompletion(0L, closedChannelException);
      }
    } else if (!readIntoCalled.compareAndSet(false, true)) {
      throw new IllegalStateException("ReadableStreamChannel cannot be read more than once");
    } else {
      callbackWrapper = new ReadIntoCallbackWrapper(callback);
      future = callbackWrapper.futureResult;
      if (size == 0) {
        callbackWrapper.invokeCallback(null);
      } else {
        writeToChannel(asyncWritableChannel);
      }
    }
    return future;
  }
Constructs a {@link ReadableStreamChannel} whose read operations return data from the provided {@code inputStream}.
<p/>
Note that {@code size} does not refer the number of bytes that is read from the {@code inputStream}. It refers
to the total size of the stream and is only used to return the size of the channel on {@link #getSize()}. The
{@code inputStream} is always read completely if {@code size} > 0.
@param inputStream the {@link InputStream} that is used to retrieve data from - on invocation of read operations.
@param size the total size of the {@code inputStream}. -1 if unknown.
@param executorService the {@link ExecutorService} that will be used to schedule blocking reads from the
                       {@code inputStream}.
ambry-commons/src/main/java/com/github/ambry/commons/InputStreamReadableStreamChannel.java