code: private void handleReads(final MemcachedNode node) throws IOException {
      Operation currentOp = node.getCurrentReadOp();
      if (currentOp instanceof TapAckOperationImpl) {
        node.removeCurrentReadOp();
        return;
      }
  
      ByteBuffer rbuf = node.getRbuf();
      final SocketChannel channel = node.getChannel();
      int read = channel.read(rbuf);
      metrics.updateHistogram(OVERALL_AVG_BYTES_READ_METRIC, read);
      if (read < 0) {
        currentOp = handleReadsWhenChannelEndOfStream(currentOp, node, rbuf);
      }
  
      while (read > 0) {
        getLogger().debug("Read %d bytes", read);
        rbuf.flip();
        while (rbuf.remaining() > 0) {
          if (currentOp == null) {
            throw new IllegalStateException("No read operation.");
          }
  
          long timeOnWire =
            System.nanoTime() - currentOp.getWriteCompleteTimestamp();
          metrics.updateHistogram(OVERALL_AVG_TIME_ON_WIRE_METRIC,
            (int)(timeOnWire / 1000));
          metrics.markMeter(OVERALL_RESPONSE_METRIC);
          synchronized(currentOp) {
            readBufferAndLogMetrics(currentOp, rbuf, node);
          }
  
          currentOp = node.getCurrentReadOp();
        }
        rbuf.clear();
        read = channel.read(rbuf);
        node.completedRead();
      }
    }
docs: Handle pending reads for the given node.
  
  @param node the node to handle reads for.
  @throws IOException can be raised during reading failures.