add PeekAt(): read at an offset past the read pointer

peekat_test.go

@@ -0,0 +1,146 @@
+package ringbuffer
+
+import (
+	"bytes"
+	"testing"
+)
+
+// PeekAt exists for a consumer that runs ahead of what it can release: bytes
+// stay buffered until something confirms them, so the read pointer sits where
+// confirmation reached and the consumer reads from somewhere past it.
+
+func TestPeekAtReadsFromTheOffset(t *testing.T) {
+	rb := New(64)
+	rb.Write([]byte("0123456789"))
+
+	p := make([]byte, 4)
+	for _, tc := range []struct {
+		off  int
+		want string
+	}{
+		{0, "0123"},
+		{3, "3456"},
+		{6, "6789"},
+	} {
+		n, err := rb.PeekAt(tc.off, p)
+		if err != nil {
+			t.Fatalf("PeekAt(%d): %v", tc.off, err)
+		}
+		if got := string(p[:n]); got != tc.want {
+			t.Errorf("PeekAt(%d) = %q, want %q", tc.off, got, tc.want)
+		}
+	}
+
+	if rb.Length() != 10 {
+		t.Errorf("Length = %d, want 10 — PeekAt must not consume", rb.Length())
+	}
+}
+
+func TestPeekAtShortAtTheEnd(t *testing.T) {
+	rb := New(64)
+	rb.Write([]byte("0123456789"))
+
+	p := make([]byte, 8)
+	n, err := rb.PeekAt(7, p)
+	if err != nil {
+		t.Fatalf("PeekAt: %v", err)
+	}
+	if got := string(p[:n]); got != "789" {
+		t.Errorf("got %q, want %q", got, "789")
+	}
+}
+
+// An offset at or past the end is the ordinary "nothing new yet" case for a
+// consumer that has taken everything, and must not look like an empty buffer.
+func TestPeekAtPastTheEndIsNotAnError(t *testing.T) {
+	rb := New(64)
+	rb.Write([]byte("0123456789"))
+
+	p := make([]byte, 8)
+	for _, off := range []int{10, 11, 1000} {
+		n, err := rb.PeekAt(off, p)
+		if n != 0 || err != nil {
+			t.Errorf("PeekAt(%d) = %d, %v; want 0, nil", off, n, err)
+		}
+	}
+	if rb.Length() != 10 {
+		t.Errorf("Length = %d, want 10", rb.Length())
+	}
+}
+
+// The offset has to be applied in ring coordinates, not slice coordinates: the
+// window it names can begin before the wrap and end after it.
+func TestPeekAtAcrossTheWrap(t *testing.T) {
+	rb := New(10)
+	rb.Write([]byte("0123456789")) // full
+	rb.Read(make([]byte, 6))       // read pointer now at 6
+	rb.Write([]byte("abcdef"))     // wraps: buffered is "6789abcdef"
+
+	if rb.Length() != 10 {
+		t.Fatalf("Length = %d, want 10", rb.Length())
+	}
+
+	p := make([]byte, 10)
+	n, err := rb.PeekAt(0, p)
+	if err != nil || string(p[:n]) != "6789abcdef" {
+		t.Fatalf("PeekAt(0) = %q, %v", p[:n], err)
+	}
+
+	// Starting before the wrap and running past it.
+	n, _ = rb.PeekAt(2, p)
+	if got := string(p[:n]); got != "89abcdef" {
+		t.Errorf("PeekAt(2) = %q, want %q", got, "89abcdef")
+	}
+	// Starting after the wrap.
+	n, _ = rb.PeekAt(6, p)
+	if got := string(p[:n]); got != "cdef" {
+		t.Errorf("PeekAt(6) = %q, want %q", got, "cdef")
+	}
+}
+
+func TestPeekAtOnAFullBuffer(t *testing.T) {
+	rb := New(8)
+	rb.Write([]byte("abcdefgh")) // exactly full: w == r and isFull
+
+	p := make([]byte, 8)
+	n, _ := rb.PeekAt(0, p)
+	if got := string(p[:n]); got != "abcdefgh" {
+		t.Errorf("PeekAt(0) on a full buffer = %q", got)
+	}
+	n, _ = rb.PeekAt(5, p)
+	if got := string(p[:n]); got != "fgh" {
+		t.Errorf("PeekAt(5) on a full buffer = %q, want %q", got, "fgh")
+	}
+}
+
+func TestPeekAtEmptyAndDegenerate(t *testing.T) {
+	rb := New(8)
+
+	p := make([]byte, 4)
+	if n, err := rb.PeekAt(0, p); n != 0 || err != ErrIsEmpty {
+		t.Errorf("PeekAt on empty = %d, %v; want 0, ErrIsEmpty", n, err)
+	}
+
+	rb.Write([]byte("abcd"))
+	if n, _ := rb.PeekAt(0, nil); n != 0 {
+		t.Errorf("PeekAt with no destination returned %d", n)
+	}
+	if n, _ := rb.PeekAt(-1, p); n != 0 {
+		t.Errorf("PeekAt with a negative offset returned %d", n)
+	}
+}
+
+// PeekAt must agree with Peek where they overlap, or a consumer switching from
+// one to the other would silently shift the stream.
+func TestPeekAtAgreesWithPeek(t *testing.T) {
+	rb := New(32)
+	rb.Write([]byte("the quick brown fox jumps"))
+
+	a := make([]byte, 32)
+	b := make([]byte, 32)
+	na, _ := rb.Peek(a)
+	nb, _ := rb.PeekAt(0, b)
+	if na != nb || !bytes.Equal(a[:na], b[:nb]) {
+		t.Errorf("Peek = %q, PeekAt(0) = %q", a[:na], b[:nb])
+	}
+}

ring_buffer.go

@@ -844,19 +844,7 @@ func (r *RingBuffer) writeByte(c byte) error {
 func (r *RingBuffer) Length() int {
 	r.mu.Lock()
 	defer r.mu.Unlock()
-
-	if r.w == r.r {
-		if r.isFull {
-			return r.size
-		}
-		return 0
-	}
-
-	if r.w > r.r {
-		return r.w - r.r
-	}
-
-	return r.size - r.r + r.w
+	return r.length()
 }
 
 // Capacity returns the size of the underlying buffer.
@@ -1091,3 +1079,73 @@ func (r *RingBuffer) peek(p []byte) (n int, err error) {
 	}
 	return n, r.readErr(true)
 }
+
+// PeekAt reads up to len(p) bytes into p, starting off bytes past the read
+// pointer, without moving it.
+//
+// The offset is for a consumer that has to run ahead of what it can release.
+// Where bytes stay buffered until something confirms them, the read pointer is
+// wherever confirmation has reached — not where reading should resume — and the
+// two can be far apart. Peek always starts at the read pointer, so such a
+// consumer would otherwise have to ask for everything from there and discard
+// the part it already has, copying the skipped prefix on every call.
+//
+// An offset at or past the end of the buffered data returns 0 and no error: it
+// means the consumer has taken everything there is, which is an ordinary thing
+// for it to ask. That is distinct from an empty buffer, which returns
+// ErrIsEmpty exactly as Peek does.
+func (r *RingBuffer) PeekAt(off int, p []byte) (n int, err error) {
+	if len(p) == 0 || off < 0 {
+		return 0, r.readErr(false)
+	}
+
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	if err := r.readErr(true); err != nil {
+		return 0, err
+	}
+
+	held := r.length()
+	if held == 0 {
+		// Nothing buffered at all, which is what Peek reports here too.
+		return 0, ErrIsEmpty
+	}
+	if off >= held {
+		// Buffered, but the consumer has already taken all of it. Not the same
+		// condition as an empty buffer, and not an error: it is what a consumer
+		// that has caught up sees every time it asks.
+		return 0, nil
+	}
+
+	n = held - off
+	if n > len(p) {
+		n = len(p)
+	}
+
+	start := r.r + off
+	if start >= r.size {
+		start -= r.size
+	}
+	if start+n <= r.size {
+		copy(p, r.buf[start:start+n])
+	} else {
+		c1 := r.size - start
+		copy(p, r.buf[start:r.size])
+		copy(p[c1:], r.buf[0:n-c1])
+	}
+	return n, r.readErr(true)
+}
+
+// length is Length without the lock, for callers that already hold it.
+func (r *RingBuffer) length() int {
+	if r.w == r.r {
+		if r.isFull {
+			return r.size
+		}
+		return 0
+	}
+	if r.w > r.r {
+		return r.w - r.r
+	}
+	return r.size - r.r + r.w
+}