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 ( ) ; } }
Handle pending reads for the given node .