add a notifications channel so clients selecting amongst many event sources don't have to build a wrapper

go.mod

@@ -1,3 +1,3 @@
-module github.com/smallnest/ringbuffer
+module github.com/Arrayscape/ringbuffer
 
 go 1.19

notify_test.go

@@ -0,0 +1,203 @@
+package ringbuffer
+
+import (
+	"io"
+	"testing"
+	"time"
+)
+
+// waitNotify reports whether a signal arrives within d.
+func waitNotify(rb *RingBuffer, d time.Duration) bool {
+	select {
+	case <-rb.Notify():
+		return true
+	case <-time.After(d):
+		return false
+	}
+}
+
+// TestNotifyOnWrite is the basic contract: a consumer parked on Notify wakes
+// when data lands.
+func TestNotifyOnWrite(t *testing.T) {
+	rb := New(64)
+
+	if waitNotify(rb, 50*time.Millisecond) {
+		t.Fatal("signalled before anything was written")
+	}
+
+	go func() {
+		time.Sleep(20 * time.Millisecond)
+		rb.Write([]byte("hello"))
+	}()
+
+	if !waitNotify(rb, 2*time.Second) {
+		t.Fatal("no signal after a write")
+	}
+}
+
+// TestNotifyWorksInNonBlockingMode — the existing wakeups are all guarded by
+// r.block, so signalling only alongside them would leave Notify silent for a
+// non-blocking buffer. It is signalled from the write path instead.
+func TestNotifyWorksInNonBlockingMode(t *testing.T) {
+	rb := New(64) // default: non-blocking
+	rb.Write([]byte("x"))
+	if !waitNotify(rb, time.Second) {
+		t.Fatal("no signal in non-blocking mode")
+	}
+}
+
+// TestNotifyCoalesces — capacity 1, so a burst of writes leaves exactly one
+// pending signal. The contract is "something changed", not a count, and a woken
+// consumer re-inspects the buffer.
+func TestNotifyCoalesces(t *testing.T) {
+	rb := New(64)
+	for i := 0; i < 10; i++ {
+		rb.Write([]byte("x"))
+	}
+
+	if !waitNotify(rb, time.Second) {
+		t.Fatal("no signal after writes")
+	}
+	if waitNotify(rb, 50*time.Millisecond) {
+		t.Error("a second signal was queued; signals must coalesce")
+	}
+}
+
+// TestNotifyOnClose — a consumer waiting for data must also be woken when there
+// will never be any, or it parks forever on a closed buffer.
+func TestNotifyOnClose(t *testing.T) {
+	rb := New(64).SetBlocking(true)
+	drain(t, rb)
+
+	go func() {
+		time.Sleep(20 * time.Millisecond)
+		rb.CloseWriter()
+	}()
+
+	if !waitNotify(rb, 2*time.Second) {
+		t.Fatal("no signal on CloseWriter; a waiting consumer would hang")
+	}
+}
+
+// TestNotifyOnCloseWithError — same, for the failure path.
+func TestNotifyOnCloseWithError(t *testing.T) {
+	rb := New(64).SetBlocking(true)
+	drain(t, rb)
+
+	go func() {
+		time.Sleep(20 * time.Millisecond)
+		rb.CloseWithError(io.ErrUnexpectedEOF)
+	}()
+
+	if !waitNotify(rb, 2*time.Second) {
+		t.Fatal("no signal on CloseWithError; a waiting consumer would hang")
+	}
+}
+
+// TestNotifyDoesNotConsume — Notify must not disturb the buffer. A consumer
+// woken by it still has to Peek to see the data, and the data must be intact.
+func TestNotifyDoesNotConsume(t *testing.T) {
+	rb := New(64)
+	rb.Write([]byte("payload"))
+	<-rb.Notify()
+
+	if got := rb.Length(); got != 7 {
+		t.Errorf("Length after Notify = %d, want 7", got)
+	}
+	p := make([]byte, 7)
+	n, err := rb.Peek(p)
+	if err != nil || string(p[:n]) != "payload" {
+		t.Errorf("Peek after Notify = %q err=%v, want %q", p[:n], err, "payload")
+	}
+}
+
+// TestPeekOnlyNeverSeesEOF documents a trap for consumers that only peek.
+//
+// readErr reports io.EOF only once the buffer is empty, so data always drains
+// before end-of-stream is announced. A consumer that never consumes therefore
+// never learns the writer is finished, and will wait forever.
+//
+// This is correct behaviour, not a defect — but a peek-driven consumer has to
+// consume something eventually, and must not treat "no EOF yet" as "the writer
+// is still alive".
+func TestPeekOnlyNeverSeesEOF(t *testing.T) {
+	rb := New(64).SetBlocking(true)
+	rb.Write([]byte("data"))
+	rb.CloseWriter()
+
+	scratch := make([]byte, 64)
+	for i := 0; i < 3; i++ {
+		n, err := rb.Peek(scratch)
+		if err == io.EOF {
+			t.Fatal("Peek reported EOF while data was still buffered")
+		}
+		if n != 4 {
+			t.Fatalf("Peek returned %d bytes, want 4", n)
+		}
+	}
+
+	// Consuming is what surfaces it.
+	rb.Read(make([]byte, 4))
+	if _, err := rb.Peek(scratch); err != io.EOF {
+		t.Errorf("after draining, Peek err = %v, want io.EOF", err)
+	}
+}
+
+// TestNotifyDrivesPeekConsumeLoop is the pattern the API exists for: wait on a
+// signal, look without consuming, send, and consume only what has been
+// confirmed. Consuming on confirmation is also what eventually surfaces EOF.
+func TestNotifyDrivesPeekConsumeLoop(t *testing.T) {
+	rb := New(1024).SetBlocking(true)
+
+	go func() {
+		for _, s := range []string{"alpha", "beta", "gamma"} {
+			rb.Write([]byte(s))
+			time.Sleep(10 * time.Millisecond)
+		}
+		rb.CloseWriter()
+	}()
+
+	var seen string
+	sent := 0
+	scratch := make([]byte, 1024)
+	discard := make([]byte, 1024)
+
+	deadline := time.After(5 * time.Second)
+	for {
+		n, err := rb.Peek(scratch)
+		if n > sent {
+			seen += string(scratch[sent:n]) // only what has not gone out yet
+			sent = n
+		}
+		if err == io.EOF {
+			break
+		}
+
+		// The client confirms everything sent, so consume it. Until this
+		// happens the buffer never empties and EOF never arrives.
+		if sent > 0 {
+			rb.Read(discard[:sent])
+			sent = 0
+			continue
+		}
+
+		select {
+		case <-rb.Notify():
+		case <-deadline:
+			t.Fatal("loop stalled waiting for a signal")
+		}
+	}
+
+	if seen != "alphabetagamma" {
+		t.Errorf("got %q, want %q", seen, "alphabetagamma")
+	}
+	if got := rb.Length(); got != 0 {
+		t.Errorf("Length = %d, want 0 after everything was confirmed", got)
+	}
+}
+
+// drain keeps a blocking buffer from wedging a test when nothing reads it.
+func drain(t *testing.T, rb *RingBuffer) {
+	t.Helper()
+	t.Cleanup(func() { rb.CloseWithError(io.ErrClosedPipe) })
+}

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])
+	}
+}