From 3ef018e457d4d8447ddc484044b3bebac947781f Mon Sep 17 00:00:00 2001 From: Tad Hunt Date: Mon, 7 Sep 2026 17:06:55 -0600 Subject: [PATCH] add PeekAt() api --- peekat_test.go | 146 +++++++++++++++++++++++++++++++++++++++++++++++++ ring_buffer.go | 84 +++++++++++++++++++++++----- 2 files changed, 217 insertions(+), 13 deletions(-) create mode 100644 peekat_test.go diff --git a/peekat_test.go b/peekat_test.go new file mode 100644 index 0000000..571fbbc --- /dev/null +++ b/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]) + } +} diff --git a/ring_buffer.go b/ring_buffer.go index 4048d0f..bd1f8d8 100644 --- a/ring_buffer.go +++ b/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 +}