int rs = Math.min(bufSize, (int) ( length - p ));
            int read = is.read(buffer, 0, rs);
            if ( read < 0 ) {
                fail("Unexpected EOF");
            }

            byte verify[] = new byte[read];
            ReadWriteTest.randBytes(r, verify);
            byte actual[] = Arrays.copyOfRange(buffer, 0, read);

            assertArrayEquals("Data matches at offset " + p, actual, verify);

    /** The next byte to read. This is always less than or equal to [upstreamPos]. */
    private var sourcePos = 0L

    /**
     * Selects where to find the bytes for a read and read them. This is one of three sources.
     *
     * ## Upstream
     *
     * In this case the current thread is assigned as the upstream reader. We read bytes from

      int off = (int) expectedSize - remaining;
      int read = in.read(bytes, off, remaining);
      if (read == -1) {
        // end of stream before reading expectedSize bytes
        // just return the bytes read so far
        return Arrays.copyOf(bytes, off);
      }
      remaining -= read;
    }

    // bytes is now full
    int b = in.read();
    if (b == -1) {
      return bytes;
    }

  public void testOpenStream() throws IOException {
    Reader reader = source.openStream();

    StringWriter writer = new StringWriter();
    char[] buf = new char[64];
    int read;
    while ((read = reader.read(buf)) != -1) {
      writer.write(buf, 0, read);
    }
    reader.close();
    writer.close();

    assertExpectedString(writer.toString());
  }

  public void testOpenBufferedStream() throws IOException {

type MonitorReaderOptions struct {
	BucketOptions
	HeaderSize int
}

// Read implements a throttled read
func (r *MonitoredReader) Read(buf []byte) (n int, err error) {
	if r.throttle == nil {
		return r.r.Read(buf)
	}
	if r.lastErr != nil {
		err = r.lastErr
		return
	}
	b := r.throttle.Burst()  // maximum available tokens

     * any.
     * @return Model
     */
    public Settings read(Reader reader) throws IOException, XmlPullParserException {
        try {
            return new Settings(delegate.read(reader));
        } catch (XMLStreamException e) {
            throw new XmlPullParserException(e.getMessage(), null, e);
        }
    }

    /**
     * Method read.
     *
     * @param in a in object.

    void close () throws SmbException;


    /**
     * Read a single byte from the current position
     * 
     * @return read byte, -1 if EOF
     * @throws SmbException
     */
    int read () throws SmbException;


    /**
     * Read into buffer from current position
     * 
     * @param b
     *            buffer
     * @return number of bytes read
     * @throws SmbException
     */

            if (peekc == '\r') {
                peekc = read();
                if (peekc == '\n') {
                    peekc = read();
                }
            } else {
                peekc = read();
            }
            if (peekc < 0) {
                ttype = TT_EOF;
                return true;
            }
            peekct = peekc < 256 ? ctype[peekc] : CT_ALPHA;

	b.ResetTimer()
	for i := 0; i < b.N; i++ {
		rb.Write(data)
		rb.Read(buf)
	}
}

func BenchmarkRingBuffer_AsyncRead(b *testing.B) {
	// Pretty useless benchmark, but it's here for completeness.
	rb := New(1024)
	data := []byte(strings.Repeat("a", 512))
	buf := make([]byte, 512)

	go func() {
		for {
			rb.Read(buf)
		}
	}()

	b.ResetTimer()
	for i := 0; i < b.N; i++ {
		rb.Write(data)

		defer deadlineconn.Close()

		// Read a line
		b := make([]byte, 12)
		_, terr = deadlineconn.Read(b)
		if terr != nil {
			t.Errorf("failed to read from client. %v", terr)
			return
		}
		received := string(b)
		if received != "message one\n" {
			t.Errorf(`server: expected: "message one\n", got: %v`, received)
			return
		}

		// Wait for more than read timeout to simulate processing.