@Override
  public synchronized int read(byte[] buf, int offset, int length) throws IOException {
    if (buf == null) {
      throw new IllegalArgumentException("buf must not be null");
    }

    if (!(offset >= 0 && length > 0 && offset < buf.length && length <= (buf.length - offset))) {
      throw new IllegalArgumentException("invalid offset and lengeth");
    }

    // Return early if closed is true; throw the saved exception if needed.
    if (closed) {
      if (exception != null) {
        throw exception;
      }
      return -1;
    }

    // Check if there is already a chunk that hasn't been exhausted; call take() if not.
    if (currentChunk == null) {
      if (currentChunkReadPos != -1) {
        throw new IllegalStateException("currentChunk is null but currentChunkReadPos is not -1");
      }

      byte[] next = null;
      try {
        if (timeoutMillis >= 0) {
          next = queue.poll(timeoutMillis, TimeUnit.MILLISECONDS);
        } else {
          next = queue.take();
        }
      } catch (InterruptedException e) {
        // Unlikely to happen.
        throw new AssertionError(e);
      }
      if (next == null) {
        closed = true;
        SocketTimeoutException timeoutException = new SocketTimeoutException();

        if (exception == null) {
          exception = timeoutException;
          // It is still possible that an endOffer(Exception) races and set exception at this point,
          // but timeoutException is still being thrown, and it's ok for subsequent reads to
          // throw the exception set by endOffer(Exception).
        }

        throw timeoutException;
      }

      // If it's the end marker, acknowledge that so that the buffer will mark itself as closed
      // for reading.
      if (next == CLOSED) {
        closed = true;
        if (exception != null) {
          throw exception;
        }
        return -1;
      }
      currentChunk = next;
      currentChunkReadPos = 0;
    }

    int available = currentChunk.length - currentChunkReadPos;
    if (length < available) {
      // Copy from the currentChunk.
      System.arraycopy(currentChunk, currentChunkReadPos, buf, offset, length);
      currentChunkReadPos += length;
      return length;
    } else {
      // Copy the entire currentChunk.
      System.arraycopy(currentChunk, currentChunkReadPos, buf, offset, available);
      currentChunk = null;
      currentChunkReadPos = -1;
      return available;
    }
  }
@ Override public synchronized int read ( byte [ ] buf , int offset , int length ) throws IOException { if ( buf == null ) { throw new IllegalArgumentException ( "buf must not be null" ) ; } if ( ! ( offset >= 0 && length > 0 && offset < buf . length && length <= ( buf . length - offset ) ) ) { throw new IllegalArgumentException ( "invalid offset and lengeth" ) ; } if ( closed ) { if ( exception != null ) { throw exception ; } return - 1 ; } if ( currentChunk == null ) { if ( currentChunkReadPos != - 1 ) { throw new IllegalStateException ( "currentChunk is null but currentChunkReadPos is not -1" ) ; } byte [ ] next = null ; try { if ( timeoutMillis >= 0 ) { next = queue . poll ( timeoutMillis , TimeUnit . MILLISECONDS ) ; } else { next = queue . take ( ) ; } } catch ( InterruptedException e ) { throw new AssertionError ( e ) ; } if ( next == null ) { closed = true ; SocketTimeoutException timeoutException = new SocketTimeoutException ( ) ; if ( exception == null ) { exception = timeoutException ; } throw timeoutException ; } if ( next == CLOSED ) { closed = true ; if ( exception != null ) { throw exception ; } return - 1 ; } currentChunk = next ; currentChunkReadPos = 0 ; } int available = currentChunk . length - currentChunkReadPos ; if ( length < available ) { System . arraycopy ( currentChunk , currentChunkReadPos , buf , offset , length ) ; currentChunkReadPos += length ; return length ; } else { System . arraycopy ( currentChunk , currentChunkReadPos , buf , offset , available ) ; currentChunk = null ; currentChunkReadPos = - 1 ; return available ; } }
Reads from the current chunk or polls a next chunk from the queue. This is synchronized to
allow reads to be used on different thread while still guaranteeing their sequentiality.