pico

chore(pubsub): add more tests open rss

created pr with 105.1 on 2026-01-25T17:03:08Z · by c8ef7d19

added 105.2 on 2026-01-27T04:11:48Z · by c8ef7d19


                      1:  603ca6e = 1:  603ca6e chore(pubsub): add more tests


                      2:  501c042 ! 2:  17e00b2 feat(pubsub): round robin

added 105.3 on 2026-01-29T01:03:33Z · by c8ef7d19


                      1:  603ca6e = 1:  603ca6e chore(pubsub): add more tests


                      2:  17e00b2 = 2:  17e00b2 feat(pubsub): round robin


                      -:  ------- > 3:  e3136bd fix(pubsub): check for eof before processing and skip empty byte reads


                      -:  ------- > 4:  9a6d19e fix: rr


                      -:  ------- > 5:  5b3f3a1 fix: sending 0 byte read

added 105.4 on 2026-02-01T17:16:43Z · by c8ef7d19


                      1:  603ca6e = 1:  603ca6e chore(pubsub): add more tests


                      2:  17e00b2 = 2:  17e00b2 feat(pubsub): round robin


                      3:  e3136bd = 3:  e3136bd fix(pubsub): check for eof before processing and skip empty byte reads


                      4:  9a6d19e = 4:  9a6d19e fix: rr


                      5:  5b3f3a1 = 5:  5b3f3a1 fix: sending 0 byte read


                      -:  ------- > 6:  4fff471 refactor: fixes


                      -:  ------- > 7:  d0dfc85 chore: SetDispatch on Broker

changed status to accepted on 2026-02-23T02:02:07Z · by c8ef7d19

cmds

checkout latest patchset:

ssh pr.pico.sh print 105 | git am -3

checkout any patchset in a patch request:

ssh pr.pico.sh print 105.[rev] | git am -3

add changes to patch request:

git format-patch main --stdout | ssh pr.pico.sh pr add 105

set PR to open (enables RSS notifications):

ssh pr.pico.sh pr open 105

set PR to draft (stops RSS notifications):

ssh pr.pico.sh pr draft 105

Patchset `105.3` on 2026-01-29T01:03:32Z · commit 9a6d19e

chore(pubsub): add more tests

Eric Bower 2026-01-25T16:38:25Z

feat(pubsub): round robin

Eric Bower 2026-01-25T16:38:25Z

fix(pubsub): check for eof before processing and skip empty byte reads

Eric Bower 2026-01-28T01:42:34Z

→ fix: rr

Eric Bower 2026-01-28T01:50:04Z

fix: sending 0 byte read

Eric Bower 2026-01-28T02:50:25Z

fix: rr

Eric Bower 2026-01-28T01:50:04Z

Semantic diff summary

0 added, 9 modified, 0 signature changed, 6 removed across 5 analyzed files

pkg/apps/pipe/cli.go

function_declaration pub modified

pkg/pubsub/channel.go

method_declaration Handle removed

pkg/pubsub/multicast_test.go

function_declaration TestMulticastSubBlock modified
function_declaration TestMulticastPubBlock modified
function_declaration TestMulticastMultSubs modified

pkg/pubsub/regression_test.go

function_declaration TestChannelMessageOrdering modified
function_declaration TestDispatcherClientDirection modified
function_declaration TestChannelConcurrentPublishes modified
function_declaration TestDispatcherEmptySubscribers modified
function_declaration TestDispatcherSingleSubscriber modified

pkg/pubsub/roundrobin_test.go

function_declaration TestRoundRobinSingleSub removed
function_declaration TestRoundRobinMultipleSubs removed
function_declaration TestRoundRobinDistribution removed
function_declaration TestRoundRobinSubscriberJoinLeave removed
function_declaration TestRoundRobinMultipleChannels removed

+0 -1 pkg/apps/pipe/cli.go #

......@@ -800,7 +800,6 @@ func (handler *CliHandler) pub(cmd *CliCmd, topic string, clientID string) error
	var bk psub.MessageDispatcher
	bk = &psub.MulticastDispatcher{}
	if *broker == "round_robin" {
		fmt.Println("BROKER ROUND ROBIN")
		bk = &psub.RoundRobinDispatcher{}
	}
	channel := psub.NewChannel(name)

+5 -0 pkg/pubsub/channel.go #

......@@ -71,6 +71,11 @@ func (c *Channel) Cleanup() {
}

func (c *Channel) Handle() {
	// If no dispatcher is set, use multicast as default
	if c.Dispatcher == nil {
		c.Dispatcher = &MulticastDispatcher{}
	}

	c.handleOnce.Do(func() {
		go func() {
			defer func() {

+5 -5 pkg/pubsub/multicast_test.go #

......@@ -66,7 +66,7 @@ func TestMulticastSubBlock(t *testing.T) {
		t.Fatalf("\norderActual:(%s)\norderExpected:(%s)", orderActual, orderExpected)
	}
	if actual.String() != expected {
		t.Fatalf("\nactual:(%s)\nexpected:(%s)", actual, expected)
		t.Fatalf("\nactual:(%s)\nexpected:(%s)", actual.String(), expected)
	}
}

......@@ -96,8 +96,8 @@ func TestMulticastPubBlock(t *testing.T) {

	go func() {
		orderActual += "sub-"
		wg.Done()
		fmt.Println(cast.Sub(context.TODO(), "2", actual, []*Channel{channel}, false))
		wg.Done()
	}()

	wg.Wait()

......@@ -106,7 +106,7 @@ func TestMulticastPubBlock(t *testing.T) {
		t.Fatalf("\norderActual:(%s)\norderExpected:(%s)", orderActual, orderExpected)
	}
	if actual.String() != expected {
		t.Fatalf("\nactual:(%s)\nexpected:(%s)", actual, expected)
		t.Fatalf("\nactual:(%s)\nexpected:(%s)", actual.String(), expected)
	}
}

......@@ -156,9 +156,9 @@ func TestMulticastMultSubs(t *testing.T) {
		t.Fatalf("\norderActual:(%s)\norderExpected:(%s)", orderActual, orderExpected)
	}
	if actual.String() != expected {
		t.Fatalf("\nactual:(%s)\nexpected:(%s)", actual, expected)
		t.Fatalf("\nactual:(%s)\nexpected:(%s)", actual.String(), expected)
	}
	if actualOther.String() != expected {
		t.Fatalf("\nactual:(%s)\nexpected:(%s)", actualOther, expected)
		t.Fatalf("\nactual:(%s)\nexpected:(%s)", actualOther.String(), expected)
	}
}

+0 -184 pkg/pubsub/regression_test.go #

......@@ -55,45 +55,6 @@ func TestChannelMessageOrdering(t *testing.T) {
			t.Error("Multicast: no messages received")
		}
	})

	// Test with RoundRobin
	t.Run("RoundRobin", func(t *testing.T) {
		rr := NewRoundRobin(slog.Default())
		buf := new(Buffer)
		channel := NewChannel(name)

		var wg sync.WaitGroup
		syncer := make(chan int)

		// Subscribe
		wg.Add(1)
		go func() {
			defer wg.Done()
			syncer <- 0
			_ = rr.Sub(context.TODO(), "sub", buf, []*Channel{channel}, false)
		}()

		<-syncer

		// Publish messages
		for i := 0; i < numMessages; i++ {
			wg.Add(1)
			idx := i
			go func() {
				defer wg.Done()
				msg := fmt.Sprintf("msg%d\n", idx)
				_ = rr.Pub(context.TODO(), "pub", &Buffer{b: *bytes.NewBufferString(msg)}, []*Channel{channel}, false)
			}()
		}

		wg.Wait()

		// Verify at least some messages were received
		content := buf.String()
		if len(content) == 0 {
			t.Error("RoundRobin: no messages received")
		}
	})
}

// TestDispatcherClientDirection verifies that both dispatchers respect client direction.

......@@ -135,41 +96,6 @@ func TestDispatcherClientDirection(t *testing.T) {
			t.Errorf("Subscriber should have received message, got: %q", subBuf.String())
		}
	})

	t.Run("RoundRobin", func(t *testing.T) {
		rr := NewRoundRobin(slog.Default())
		pubBuf := new(Buffer)
		subBuf := new(Buffer)
		channel := NewChannel(name)

		var wg sync.WaitGroup

		// Publisher (input only)
		wg.Add(1)
		go func() {
			defer wg.Done()
			_ = rr.Pub(context.TODO(), "pub", &Buffer{b: *bytes.NewBufferString("test")}, []*Channel{channel}, false)
		}()

		// Subscriber (output only)
		wg.Add(1)
		go func() {
			defer wg.Done()
			_ = rr.Sub(context.TODO(), "sub", subBuf, []*Channel{channel}, false)
		}()

		wg.Wait()

		// Publisher should not receive the message
		if pubBuf.String() != "" {
			t.Errorf("Publisher received message: %q", pubBuf.String())
		}

		// Subscriber should receive it
		if subBuf.String() != "test" {
			t.Errorf("Subscriber should have received message, got: %q", subBuf.String())
		}
	})
}

// TestChannelConcurrentPublishes verifies that concurrent publishes don't cause races or data loss.

......@@ -232,64 +158,6 @@ func TestChannelConcurrentPublishes(t *testing.T) {
			t.Errorf("Expected %d publishes to complete, got %d", totalExpectedMessages, pubCount)
		}
	})

	t.Run("RoundRobin", func(t *testing.T) {
		rr := NewRoundRobin(slog.Default())
		buffers := make([]*Buffer, numSubscribers)
		for i := range buffers {
			buffers[i] = new(Buffer)
		}
		channel := NewChannel(name)

		var wg sync.WaitGroup

		// Subscribe
		for i := range buffers {
			wg.Add(1)
			idx := i
			go func() {
				defer wg.Done()
				_ = rr.Sub(context.TODO(), fmt.Sprintf("sub-%d", idx), buffers[idx], []*Channel{channel}, false)
			}()
		}
		time.Sleep(100 * time.Millisecond)

		// Concurrent publishers
		pubCount := int32(0)
		for p := 0; p < numPublishers; p++ {
			pubID := p
			for m := 0; m < msgsPerPublisher; m++ {
				wg.Add(1)
				msgNum := m
				go func() {
					defer wg.Done()
					msg := fmt.Sprintf("pub%d-msg%d\n", pubID, msgNum)
					_ = rr.Pub(context.TODO(), fmt.Sprintf("pub-%d", pubID), &Buffer{b: *bytes.NewBufferString(msg)}, []*Channel{channel}, false)
					atomic.AddInt32(&pubCount, 1)
				}()
			}
		}

		wg.Wait()

		// Verify all messages distributed (one to each subscriber per round-robin cycle)
		// Allow for some timing variance - expect at least 90% of messages
		totalExpectedMessages := numPublishers * msgsPerPublisher
		totalDelivered := int32(0)
		for _, buf := range buffers {
			messageCount := bytes.Count([]byte(buf.String()), []byte("\n"))
			totalDelivered += int32(messageCount)
		}
		minExpected := int32(float32(totalExpectedMessages) * 0.9)
		if totalDelivered < minExpected {
			t.Errorf("Expected at least %d messages, got %d", minExpected, totalDelivered)
		}

		// Verify all publishes completed
		if pubCount != int32(totalExpectedMessages) {
			t.Errorf("Expected %d publishes to complete, got %d", totalExpectedMessages, pubCount)
		}
	})
}

// TestDispatcherEmptySubscribers verifies that dispatchers handle empty subscriber set without panic.

......@@ -317,28 +185,6 @@ func TestDispatcherEmptySubscribers(t *testing.T) {
		wg.Wait()
		t.Log("Multicast handled empty subscribers correctly")
	})

	t.Run("RoundRobin", func(t *testing.T) {
		rr := NewRoundRobin(slog.Default())
		channel := NewChannel(name)

		var wg sync.WaitGroup

		// Publish with no subscribers (should not panic)
		wg.Add(1)
		go func() {
			defer wg.Done()
			defer func() {
				if r := recover(); r != nil {
					t.Errorf("RoundRobin panicked with no subscribers: %v", r)
				}
			}()
			_ = rr.Pub(context.TODO(), "pub", &Buffer{b: *bytes.NewBufferString("test")}, []*Channel{channel}, false)
		}()

		wg.Wait()
		t.Log("RoundRobin handled empty subscribers correctly")
	})
}

// TestDispatcherSingleSubscriber verifies that both dispatchers work correctly with one subscriber.

......@@ -375,34 +221,4 @@ func TestDispatcherSingleSubscriber(t *testing.T) {
			t.Errorf("Multicast with single subscriber: expected %q, got %q", message, buf.String())
		}
	})

	t.Run("RoundRobin", func(t *testing.T) {
		rr := NewRoundRobin(slog.Default())
		buf := new(Buffer)
		channel := NewChannel(name)

		var wg sync.WaitGroup

		// Subscribe
		wg.Add(1)
		go func() {
			defer wg.Done()
			_ = rr.Sub(context.TODO(), "sub", buf, []*Channel{channel}, false)
		}()

		time.Sleep(100 * time.Millisecond)

		// Publish
		wg.Add(1)
		go func() {
			defer wg.Done()
			_ = rr.Pub(context.TODO(), "pub", &Buffer{b: *bytes.NewBufferString(message)}, []*Channel{channel}, false)
		}()

		wg.Wait()

		if buf.String() != message {
			t.Errorf("RoundRobin with single subscriber: expected %q, got %q", message, buf.String())
		}
	})
}

+0 -354 pkg/pubsub/roundrobin_test.go #

......@@ -1,354 +0,0 @@
package pubsub

import (
	"bytes"
	"context"
	"fmt"
	"log/slog"
	"sync"
	"testing"
	"time"
)

func TestRoundRobinSingleSub(t *testing.T) {
	// Single publisher, single subscriber
	// Should work like normal pub/sub
	orderActual := ""
	orderExpected := "sub-pub-"
	actual := new(Buffer)
	expected := "some test data"
	name := "test-channel"
	syncer := make(chan int)

	rr := NewRoundRobin(slog.Default())

	var wg sync.WaitGroup
	wg.Add(2)

	channel := NewChannel(name)

	go func() {
		orderActual += "sub-"
		syncer <- 0
		fmt.Println(rr.Sub(context.TODO(), "1", actual, []*Channel{channel}, false))
		wg.Done()
	}()

	<-syncer

	go func() {
		orderActual += "pub-"
		fmt.Println(rr.Pub(context.TODO(), "2", &Buffer{b: *bytes.NewBufferString(expected)}, []*Channel{channel}, true))
		wg.Done()
	}()

	wg.Wait()

	if orderActual != orderExpected {
		t.Fatalf("\norderActual:(%s)\norderExpected:(%s)", orderActual, orderExpected)
	}
	if actual.String() != expected {
		t.Fatalf("\nactual:(%s)\nexpected:(%s)", actual.String(), expected)
	}
}

func TestRoundRobinMultipleSubs(t *testing.T) {
	// Single publisher, multiple subscribers
	// Verify round-robin distributes across subscribers
	name := "test-channel"

	rr := NewRoundRobin(slog.Default())

	buffers := []*Buffer{new(Buffer), new(Buffer), new(Buffer)}
	channel := NewChannel(name)

	var wg sync.WaitGroup

	// Subscribe three clients sequentially with sync point
	syncer := make(chan int, 3)
	for i := range buffers {
		idx := i
		wg.Add(1)
		go func() {
			defer wg.Done()
			clientID := fmt.Sprintf("sub-%d", idx)
			_ = rr.Sub(context.TODO(), clientID, buffers[idx], []*Channel{channel}, false)
		}()
		syncer <- i
	}

	// Wait for all subscribers to connect
	for i := 0; i < 3; i++ {
		<-syncer
	}
	time.Sleep(200 * time.Millisecond)

	// Publish many messages to verify distribution
	numMsgs := 9
	for i := 0; i < numMsgs; i++ {
		wg.Add(1)
		idx := i
		go func() {
			defer wg.Done()
			msg := fmt.Sprintf("msg%d\n", idx)
			_ = rr.Pub(context.TODO(), "pub", &Buffer{b: *bytes.NewBufferString(msg)}, []*Channel{channel}, false)
		}()
	}

	wg.Wait()

	// Verify that messages were distributed across all subscribers
	for i, buf := range buffers {
		content := buf.String()
		t.Logf("sub-%d received %d bytes", i, len(content))
		if len(content) == 0 {
			t.Logf("WARNING: sub-%d received no messages", i)
		}
	}

	// At least one subscriber should have received messages
	totalLen := 0
	for _, buf := range buffers {
		totalLen += len(buf.String())
	}
	if totalLen == 0 {
		t.Fatal("No messages were delivered to any subscriber")
	}
}

func TestRoundRobinDistribution(t *testing.T) {
	// Verify that messages are distributed evenly across subscribers
	expected := "msg"
	name := "test-channel"
	numSubs := 3
	numMessages := 9

	rr := NewRoundRobin(slog.Default())

	buffers := make([]*Buffer, numSubs)
	for i := 0; i < numSubs; i++ {
		buffers[i] = new(Buffer)
	}
	channels := []*Channel{NewChannel(name)}

	var wg sync.WaitGroup

	// Subscribe clients
	for i := 0; i < numSubs; i++ {
		wg.Add(1)
		idx := i
		go func() {
			defer wg.Done()
			clientID := fmt.Sprintf("sub-%d", idx)
			_ = rr.Sub(context.TODO(), clientID, buffers[idx], channels, false)
		}()
	}

	time.Sleep(100 * time.Millisecond)

	// Publish multiple messages
	for i := 0; i < numMessages; i++ {
		wg.Add(1)
		msgIdx := i
		go func() {
			defer wg.Done()
			msg := fmt.Sprintf("%s%d\n", expected, msgIdx)
			_ = rr.Pub(context.TODO(), "pub", &Buffer{b: *bytes.NewBufferString(msg)}, channels, false)
		}()
	}

	wg.Wait()

	// Count messages per subscriber
	msgCounts := make(map[int]int)
	for i, buf := range buffers {
		// Count occurrences of "msg" in the buffer
		content := buf.String()
		count := 0
		for j := 0; j < numMessages; j++ {
			marker := fmt.Sprintf("msg%d", j)
			if bytes.Contains([]byte(content), []byte(marker)) {
				count++
			}
		}
		msgCounts[i] = count
		t.Logf("sub-%d received %d messages", i, count)
	}

	// Verify relatively even distribution (within 1 message difference due to concurrency)
	minCount := msgCounts[0]
	maxCount := msgCounts[0]
	for i := 1; i < numSubs; i++ {
		if msgCounts[i] < minCount {
			minCount = msgCounts[i]
		}
		if msgCounts[i] > maxCount {
			maxCount = msgCounts[i]
		}
	}

	if maxCount-minCount > 2 {
		t.Fatalf("Uneven distribution: min=%d, max=%d, difference=%d", minCount, maxCount, maxCount-minCount)
	}
}

func TestRoundRobinSubscriberJoinLeave(t *testing.T) {
	// Test behavior when subscribers join and leave mid-stream
	// Verify the broker gracefully handles subscriber changes
	name := "test-channel"

	rr := NewRoundRobin(slog.Default())
	channel := NewChannel(name)

	buf1 := new(Buffer)
	buf2 := new(Buffer)
	buf3 := new(Buffer)

	var wg sync.WaitGroup

	ctx1, cancel1 := context.WithCancel(context.Background())
	ctx2, cancel2 := context.WithCancel(context.Background())

	// Start with 2 subscribers
	wg.Add(2)
	go func() {
		defer wg.Done()
		_ = rr.Sub(ctx1, "sub-1", buf1, []*Channel{channel}, false)
	}()
	go func() {
		defer wg.Done()
		_ = rr.Sub(ctx2, "sub-2", buf2, []*Channel{channel}, false)
	}()

	time.Sleep(200 * time.Millisecond)

	// Publish some messages with 2 subscribers
	for i := 0; i < 2; i++ {
		wg.Add(1)
		idx := i
		go func() {
			defer wg.Done()
			msg := fmt.Sprintf("msg%d\n", idx)
			_ = rr.Pub(context.TODO(), "pub", &Buffer{b: *bytes.NewBufferString(msg)}, []*Channel{channel}, false)
		}()
	}
	time.Sleep(100 * time.Millisecond)

	// Remove sub-1
	cancel1()
	time.Sleep(200 * time.Millisecond)

	// Add sub-3
	ctx3, cancel3 := context.WithCancel(context.Background())
	wg.Add(1)
	go func() {
		defer wg.Done()
		_ = rr.Sub(ctx3, "sub-3", buf3, []*Channel{channel}, false)
	}()

	time.Sleep(200 * time.Millisecond)

	// Publish more messages with different subscriber set
	for i := 2; i < 4; i++ {
		wg.Add(1)
		idx := i
		go func() {
			defer wg.Done()
			msg := fmt.Sprintf("msg%d\n", idx)
			_ = rr.Pub(context.TODO(), "pub", &Buffer{b: *bytes.NewBufferString(msg)}, []*Channel{channel}, false)
		}()
	}

	wg.Wait()
	cancel2()
	cancel3()

	t.Logf("sub-1: %d bytes", len(buf1.String()))
	t.Logf("sub-2: %d bytes", len(buf2.String()))
	t.Logf("sub-3: %d bytes", len(buf3.String()))

	// Verify that messages were delivered (exact distribution depends on timing)
	totalLen := len(buf1.String()) + len(buf2.String()) + len(buf3.String())
	if totalLen == 0 {
		t.Fatal("No messages were delivered after subscriber changes")
	}
}

func TestRoundRobinMultipleChannels(t *testing.T) {
	// Test that each channel maintains independent round-robin state
	rr := NewRoundRobin(slog.Default())

	ch1 := NewChannel("topic-1")
	ch2 := NewChannel("topic-2")

	buf1ch1 := new(Buffer)
	buf2ch1 := new(Buffer)
	buf1ch2 := new(Buffer)
	buf2ch2 := new(Buffer)

	var wg sync.WaitGroup

	// Subscribe to channel 1
	wg.Add(2)
	go func() {
		defer wg.Done()
		_ = rr.Sub(context.TODO(), "sub-1-ch1", buf1ch1, []*Channel{ch1}, false)
	}()
	go func() {
		defer wg.Done()
		_ = rr.Sub(context.TODO(), "sub-2-ch1", buf2ch1, []*Channel{ch1}, false)
	}()

	// Subscribe to channel 2
	wg.Add(2)
	go func() {
		defer wg.Done()
		_ = rr.Sub(context.TODO(), "sub-1-ch2", buf1ch2, []*Channel{ch2}, false)
	}()
	go func() {
		defer wg.Done()
		_ = rr.Sub(context.TODO(), "sub-2-ch2", buf2ch2, []*Channel{ch2}, false)
	}()

	time.Sleep(100 * time.Millisecond)

	// Publish to channel 1
	wg.Add(2)
	for i := 0; i < 2; i++ {
		idx := i
		go func() {
			defer wg.Done()
			msg := fmt.Sprintf("ch1-msg%d\n", idx)
			_ = rr.Pub(context.TODO(), "pub-1", &Buffer{b: *bytes.NewBufferString(msg)}, []*Channel{ch1}, false)
		}()
	}

	// Publish to channel 2
	wg.Add(2)
	for i := 0; i < 2; i++ {
		idx := i
		go func() {
			defer wg.Done()
			msg := fmt.Sprintf("ch2-msg%d\n", idx)
			_ = rr.Pub(context.TODO(), "pub-2", &Buffer{b: *bytes.NewBufferString(msg)}, []*Channel{ch2}, false)
		}()
	}

	wg.Wait()

	t.Logf("ch1-buf1: %s", buf1ch1.String())
	t.Logf("ch1-buf2: %s", buf2ch1.String())
	t.Logf("ch2-buf1: %s", buf1ch2.String())
	t.Logf("ch2-buf2: %s", buf2ch2.String())

	// Both channels should have distributed messages independently
	ch1Total := len(buf1ch1.String()) + len(buf2ch1.String())
	ch2Total := len(buf1ch2.String()) + len(buf2ch2.String())

	if ch1Total == 0 {
		t.Fatal("Channel 1 should have received messages")
	}
	if ch2Total == 0 {
		t.Fatal("Channel 2 should have received messages")
	}
}

...	...		@@ -106,7 +106,7 @@ func TestMulticastPubBlock(t *testing.T) {
106	106		t.Fatalf("\norderActual:(%s)\norderExpected:(%s)", orderActual, orderExpected)
107	107		}
108	108		if actual.String() != expected {
109		-	t.Fatalf("\nactual:(%s)\nexpected:(%s)", actual, expected)
	109	+	t.Fatalf("\nactual:(%s)\nexpected:(%s)", actual.String(), expected)
110	110		}
111	111		}
112	112

...	...		@@ -135,41 +96,6 @@ func TestDispatcherClientDirection(t *testing.T) {
135	96		t.Errorf("Subscriber should have received message, got: %q", subBuf.String())
136	97		}
137	98		})
138		-
139		-	t.Run("RoundRobin", func(t *testing.T) {
140		-	rr := NewRoundRobin(slog.Default())
141		-	pubBuf := new(Buffer)
142		-	subBuf := new(Buffer)
143		-	channel := NewChannel(name)
144		-
145		-	var wg sync.WaitGroup
146		-
147		-	// Publisher (input only)
148		-	wg.Add(1)
149		-	go func() {
150		-	defer wg.Done()
151		-	_ = rr.Pub(context.TODO(), "pub", &Buffer{b: bytes.NewBufferString("test")}, []Channel{channel}, false)
152		-	}()
153		-
154		-	// Subscriber (output only)
155		-	wg.Add(1)
156		-	go func() {
157		-	defer wg.Done()
158		-	_ = rr.Sub(context.TODO(), "sub", subBuf, []*Channel{channel}, false)
159		-	}()
160		-
161		-	wg.Wait()
162		-
163		-	// Publisher should not receive the message
164		-	if pubBuf.String() != "" {
165		-	t.Errorf("Publisher received message: %q", pubBuf.String())
166		-	}
167		-
168		-	// Subscriber should receive it
169		-	if subBuf.String() != "test" {
170		-	t.Errorf("Subscriber should have received message, got: %q", subBuf.String())
171		-	}
172		-	})
173	99		}
174	100
175	101		// TestChannelConcurrentPublishes verifies that concurrent publishes don't cause races or data loss.

...	...		@@ -232,64 +158,6 @@ func TestChannelConcurrentPublishes(t *testing.T) {
232	158		t.Errorf("Expected %d publishes to complete, got %d", totalExpectedMessages, pubCount)
233	159		}
234	160		})
235		-
236		-	t.Run("RoundRobin", func(t *testing.T) {
237		-	rr := NewRoundRobin(slog.Default())
238		-	buffers := make([]*Buffer, numSubscribers)
239		-	for i := range buffers {
240		-	buffers[i] = new(Buffer)
241		-	}
242		-	channel := NewChannel(name)
243		-
244		-	var wg sync.WaitGroup
245		-
246		-	// Subscribe
247		-	for i := range buffers {
248		-	wg.Add(1)
249		-	idx := i
250		-	go func() {
251		-	defer wg.Done()
252		-	_ = rr.Sub(context.TODO(), fmt.Sprintf("sub-%d", idx), buffers[idx], []*Channel{channel}, false)
253		-	}()
254		-	}
255		-	time.Sleep(100 * time.Millisecond)
256		-
257		-	// Concurrent publishers
258		-	pubCount := int32(0)
259		-	for p := 0; p < numPublishers; p++ {
260		-	pubID := p
261		-	for m := 0; m < msgsPerPublisher; m++ {
262		-	wg.Add(1)
263		-	msgNum := m
264		-	go func() {
265		-	defer wg.Done()
266		-	msg := fmt.Sprintf("pub%d-msg%d\n", pubID, msgNum)
267		-	_ = rr.Pub(context.TODO(), fmt.Sprintf("pub-%d", pubID), &Buffer{b: bytes.NewBufferString(msg)}, []Channel{channel}, false)
268		-	atomic.AddInt32(&pubCount, 1)
269		-	}()
270		-	}
271		-	}
272		-
273		-	wg.Wait()
274		-
275		-	// Verify all messages distributed (one to each subscriber per round-robin cycle)
276		-	// Allow for some timing variance - expect at least 90% of messages
277		-	totalExpectedMessages := numPublishers * msgsPerPublisher
278		-	totalDelivered := int32(0)
279		-	for _, buf := range buffers {
280		-	messageCount := bytes.Count([]byte(buf.String()), []byte("\n"))
281		-	totalDelivered += int32(messageCount)
282		-	}
283		-	minExpected := int32(float32(totalExpectedMessages) * 0.9)
284		-	if totalDelivered < minExpected {
285		-	t.Errorf("Expected at least %d messages, got %d", minExpected, totalDelivered)
286		-	}
287		-
288		-	// Verify all publishes completed
289		-	if pubCount != int32(totalExpectedMessages) {
290		-	t.Errorf("Expected %d publishes to complete, got %d", totalExpectedMessages, pubCount)
291		-	}
292		-	})
293	161		}
294	162
295	163		// TestDispatcherEmptySubscribers verifies that dispatchers handle empty subscriber set without panic.

...	...		@@ -317,28 +185,6 @@ func TestDispatcherEmptySubscribers(t *testing.T) {
317	185		wg.Wait()
318	186		t.Log("Multicast handled empty subscribers correctly")
319	187		})
320		-
321		-	t.Run("RoundRobin", func(t *testing.T) {
322		-	rr := NewRoundRobin(slog.Default())
323		-	channel := NewChannel(name)
324		-
325		-	var wg sync.WaitGroup
326		-
327		-	// Publish with no subscribers (should not panic)
328		-	wg.Add(1)
329		-	go func() {
330		-	defer wg.Done()
331		-	defer func() {
332		-	if r := recover(); r != nil {
333		-	t.Errorf("RoundRobin panicked with no subscribers: %v", r)
334		-	}
335		-	}()
336		-	_ = rr.Pub(context.TODO(), "pub", &Buffer{b: bytes.NewBufferString("test")}, []Channel{channel}, false)
337		-	}()
338		-
339		-	wg.Wait()
340		-	t.Log("RoundRobin handled empty subscribers correctly")
341		-	})
342	188		}
343	189
344	190		// TestDispatcherSingleSubscriber verifies that both dispatchers work correctly with one subscriber.

...	...		@@ -375,34 +221,4 @@ func TestDispatcherSingleSubscriber(t *testing.T) {
375	221		t.Errorf("Multicast with single subscriber: expected %q, got %q", message, buf.String())
376	222		}
377	223		})
378		-
379		-	t.Run("RoundRobin", func(t *testing.T) {
380		-	rr := NewRoundRobin(slog.Default())
381		-	buf := new(Buffer)
382		-	channel := NewChannel(name)
383		-
384		-	var wg sync.WaitGroup
385		-
386		-	// Subscribe
387		-	wg.Add(1)
388		-	go func() {
389		-	defer wg.Done()
390		-	_ = rr.Sub(context.TODO(), "sub", buf, []*Channel{channel}, false)
391		-	}()
392		-
393		-	time.Sleep(100 * time.Millisecond)
394		-
395		-	// Publish
396		-	wg.Add(1)
397		-	go func() {
398		-	defer wg.Done()
399		-	_ = rr.Pub(context.TODO(), "pub", &Buffer{b: bytes.NewBufferString(message)}, []Channel{channel}, false)
400		-	}()
401		-
402		-	wg.Wait()
403		-
404		-	if buf.String() != message {
405		-	t.Errorf("RoundRobin with single subscriber: expected %q, got %q", message, buf.String())
406		-	}
407		-	})
408	224		}

...	...		@@ -800,7 +800,6 @@ func (handler CliHandler) pub(cmd CliCmd, topic string, clientID string) error
800	800		var bk psub.MessageDispatcher
801	801		bk = &psub.MulticastDispatcher{}
802	802		if *broker == "round_robin" {
803		-	fmt.Println("BROKER ROUND ROBIN")
804	803		bk = &psub.RoundRobinDispatcher{}
805	804		}
806	805		channel := psub.NewChannel(name)

...	...		@@ -71,6 +71,11 @@ func (c *Channel) Cleanup() {
71	71		}
72	72
73	73		func (c *Channel) Handle() {
	74	+	// If no dispatcher is set, use multicast as default
	75	+	if c.Dispatcher == nil {
	76	+	c.Dispatcher = &MulticastDispatcher{}
	77	+	}
	78	+
74	79		c.handleOnce.Do(func() {
75	80		go func() {
76	81		defer func() {

...	...		@@ -66,7 +66,7 @@ func TestMulticastSubBlock(t *testing.T) {
66	66		t.Fatalf("\norderActual:(%s)\norderExpected:(%s)", orderActual, orderExpected)
67	67		}
68	68		if actual.String() != expected {
69		-	t.Fatalf("\nactual:(%s)\nexpected:(%s)", actual, expected)
	69	+	t.Fatalf("\nactual:(%s)\nexpected:(%s)", actual.String(), expected)
70	70		}
71	71		}
72	72

...	...		@@ -96,8 +96,8 @@ func TestMulticastPubBlock(t *testing.T) {
96	96
97	97		go func() {
98	98		orderActual += "sub-"
99		-	wg.Done()
100	99		fmt.Println(cast.Sub(context.TODO(), "2", actual, []*Channel{channel}, false))
	100	+	wg.Done()
101	101		}()
102	102
103	103		wg.Wait()

...	...		@@ -156,9 +156,9 @@ func TestMulticastMultSubs(t *testing.T) {
156	156		t.Fatalf("\norderActual:(%s)\norderExpected:(%s)", orderActual, orderExpected)
157	157		}
158	158		if actual.String() != expected {
159		-	t.Fatalf("\nactual:(%s)\nexpected:(%s)", actual, expected)
	159	+	t.Fatalf("\nactual:(%s)\nexpected:(%s)", actual.String(), expected)
160	160		}
161	161		if actualOther.String() != expected {
162		-	t.Fatalf("\nactual:(%s)\nexpected:(%s)", actualOther, expected)
	162	+	t.Fatalf("\nactual:(%s)\nexpected:(%s)", actualOther.String(), expected)
163	163		}
164	164		}

...	...		@@ -55,45 +55,6 @@ func TestChannelMessageOrdering(t *testing.T) {
55	55		t.Error("Multicast: no messages received")
56	56		}
57	57		})
58		-
59		-	// Test with RoundRobin
60		-	t.Run("RoundRobin", func(t *testing.T) {
61		-	rr := NewRoundRobin(slog.Default())
62		-	buf := new(Buffer)
63		-	channel := NewChannel(name)
64		-
65		-	var wg sync.WaitGroup
66		-	syncer := make(chan int)
67		-
68		-	// Subscribe
69		-	wg.Add(1)
70		-	go func() {
71		-	defer wg.Done()
72		-	syncer <- 0
73		-	_ = rr.Sub(context.TODO(), "sub", buf, []*Channel{channel}, false)
74		-	}()
75		-
76		-	<-syncer
77		-
78		-	// Publish messages
79		-	for i := 0; i < numMessages; i++ {
80		-	wg.Add(1)
81		-	idx := i
82		-	go func() {
83		-	defer wg.Done()
84		-	msg := fmt.Sprintf("msg%d\n", idx)
85		-	_ = rr.Pub(context.TODO(), "pub", &Buffer{b: bytes.NewBufferString(msg)}, []Channel{channel}, false)
86		-	}()
87		-	}
88		-
89		-	wg.Wait()
90		-
91		-	// Verify at least some messages were received
92		-	content := buf.String()
93		-	if len(content) == 0 {
94		-	t.Error("RoundRobin: no messages received")
95		-	}
96		-	})
97	58		}
98	59
99	60		// TestDispatcherClientDirection verifies that both dispatchers respect client direction.

...	...		@@ -1,354 +0,0 @@
1		-	package pubsub
2		-
3		-	import (
4		-	"bytes"
5		-	"context"
6		-	"fmt"
7		-	"log/slog"
8		-	"sync"
9		-	"testing"
10		-	"time"
11		-	)
12		-
13		-	func TestRoundRobinSingleSub(t *testing.T) {
14		-	// Single publisher, single subscriber
15		-	// Should work like normal pub/sub
16		-	orderActual := ""
17		-	orderExpected := "sub-pub-"
18		-	actual := new(Buffer)
19		-	expected := "some test data"
20		-	name := "test-channel"
21		-	syncer := make(chan int)
22		-
23		-	rr := NewRoundRobin(slog.Default())
24		-
25		-	var wg sync.WaitGroup
26		-	wg.Add(2)
27		-
28		-	channel := NewChannel(name)
29		-
30		-	go func() {
31		-	orderActual += "sub-"
32		-	syncer <- 0
33		-	fmt.Println(rr.Sub(context.TODO(), "1", actual, []*Channel{channel}, false))
34		-	wg.Done()
35		-	}()
36		-
37		-	<-syncer
38		-
39		-	go func() {
40		-	orderActual += "pub-"
41		-	fmt.Println(rr.Pub(context.TODO(), "2", &Buffer{b: bytes.NewBufferString(expected)}, []Channel{channel}, true))
42		-	wg.Done()
43		-	}()
44		-
45		-	wg.Wait()
46		-
47		-	if orderActual != orderExpected {
48		-	t.Fatalf("\norderActual:(%s)\norderExpected:(%s)", orderActual, orderExpected)
49		-	}
50		-	if actual.String() != expected {
51		-	t.Fatalf("\nactual:(%s)\nexpected:(%s)", actual.String(), expected)
52		-	}
53		-	}
54		-
55		-	func TestRoundRobinMultipleSubs(t *testing.T) {
56		-	// Single publisher, multiple subscribers
57		-	// Verify round-robin distributes across subscribers
58		-	name := "test-channel"
59		-
60		-	rr := NewRoundRobin(slog.Default())
61		-
62		-	buffers := []*Buffer{new(Buffer), new(Buffer), new(Buffer)}
63		-	channel := NewChannel(name)
64		-
65		-	var wg sync.WaitGroup
66		-
67		-	// Subscribe three clients sequentially with sync point
68		-	syncer := make(chan int, 3)
69		-	for i := range buffers {
70		-	idx := i
71		-	wg.Add(1)
72		-	go func() {
73		-	defer wg.Done()
74		-	clientID := fmt.Sprintf("sub-%d", idx)
75		-	_ = rr.Sub(context.TODO(), clientID, buffers[idx], []*Channel{channel}, false)
76		-	}()
77		-	syncer <- i
78		-	}
79		-
80		-	// Wait for all subscribers to connect
81		-	for i := 0; i < 3; i++ {
82		-	<-syncer
83		-	}
84		-	time.Sleep(200 * time.Millisecond)
85		-
86		-	// Publish many messages to verify distribution
87		-	numMsgs := 9
88		-	for i := 0; i < numMsgs; i++ {
89		-	wg.Add(1)
90		-	idx := i
91		-	go func() {
92		-	defer wg.Done()
93		-	msg := fmt.Sprintf("msg%d\n", idx)
94		-	_ = rr.Pub(context.TODO(), "pub", &Buffer{b: bytes.NewBufferString(msg)}, []Channel{channel}, false)
95		-	}()
96		-	}
97		-
98		-	wg.Wait()
99		-
100		-	// Verify that messages were distributed across all subscribers
101		-	for i, buf := range buffers {
102		-	content := buf.String()
103		-	t.Logf("sub-%d received %d bytes", i, len(content))
104		-	if len(content) == 0 {
105		-	t.Logf("WARNING: sub-%d received no messages", i)
106		-	}
107		-	}
108		-
109		-	// At least one subscriber should have received messages
110		-	totalLen := 0
111		-	for _, buf := range buffers {
112		-	totalLen += len(buf.String())
113		-	}
114		-	if totalLen == 0 {
115		-	t.Fatal("No messages were delivered to any subscriber")
116		-	}
117		-	}
118		-
119		-	func TestRoundRobinDistribution(t *testing.T) {
120		-	// Verify that messages are distributed evenly across subscribers
121		-	expected := "msg"
122		-	name := "test-channel"
123		-	numSubs := 3
124		-	numMessages := 9
125		-
126		-	rr := NewRoundRobin(slog.Default())
127		-
128		-	buffers := make([]*Buffer, numSubs)
129		-	for i := 0; i < numSubs; i++ {
130		-	buffers[i] = new(Buffer)
131		-	}
132		-	channels := []*Channel{NewChannel(name)}
133		-
134		-	var wg sync.WaitGroup
135		-
136		-	// Subscribe clients
137		-	for i := 0; i < numSubs; i++ {
138		-	wg.Add(1)
139		-	idx := i
140		-	go func() {
141		-	defer wg.Done()
142		-	clientID := fmt.Sprintf("sub-%d", idx)
143		-	_ = rr.Sub(context.TODO(), clientID, buffers[idx], channels, false)
144		-	}()
145		-	}
146		-
147		-	time.Sleep(100 * time.Millisecond)
148		-
149		-	// Publish multiple messages
150		-	for i := 0; i < numMessages; i++ {
151		-	wg.Add(1)
152		-	msgIdx := i
153		-	go func() {
154		-	defer wg.Done()
155		-	msg := fmt.Sprintf("%s%d\n", expected, msgIdx)
156		-	_ = rr.Pub(context.TODO(), "pub", &Buffer{b: *bytes.NewBufferString(msg)}, channels, false)
157		-	}()
158		-	}
159		-
160		-	wg.Wait()
161		-
162		-	// Count messages per subscriber
163		-	msgCounts := make(map[int]int)
164		-	for i, buf := range buffers {
165		-	// Count occurrences of "msg" in the buffer
166		-	content := buf.String()
167		-	count := 0
168		-	for j := 0; j < numMessages; j++ {
169		-	marker := fmt.Sprintf("msg%d", j)
170		-	if bytes.Contains([]byte(content), []byte(marker)) {
171		-	count++
172		-	}
173		-	}
174		-	msgCounts[i] = count
175		-	t.Logf("sub-%d received %d messages", i, count)
176		-	}
177		-
178		-	// Verify relatively even distribution (within 1 message difference due to concurrency)
179		-	minCount := msgCounts[0]
180		-	maxCount := msgCounts[0]
181		-	for i := 1; i < numSubs; i++ {
182		-	if msgCounts[i] < minCount {
183		-	minCount = msgCounts[i]
184		-	}
185		-	if msgCounts[i] > maxCount {
186		-	maxCount = msgCounts[i]
187		-	}
188		-	}
189		-
190		-	if maxCount-minCount > 2 {
191		-	t.Fatalf("Uneven distribution: min=%d, max=%d, difference=%d", minCount, maxCount, maxCount-minCount)
192		-	}
193		-	}
194		-
195		-	func TestRoundRobinSubscriberJoinLeave(t *testing.T) {
196		-	// Test behavior when subscribers join and leave mid-stream
197		-	// Verify the broker gracefully handles subscriber changes
198		-	name := "test-channel"
199		-
200		-	rr := NewRoundRobin(slog.Default())
201		-	channel := NewChannel(name)
202		-
203		-	buf1 := new(Buffer)
204		-	buf2 := new(Buffer)
205		-	buf3 := new(Buffer)
206		-
207		-	var wg sync.WaitGroup
208		-
209		-	ctx1, cancel1 := context.WithCancel(context.Background())
210		-	ctx2, cancel2 := context.WithCancel(context.Background())
211		-
212		-	// Start with 2 subscribers
213		-	wg.Add(2)
214		-	go func() {
215		-	defer wg.Done()
216		-	_ = rr.Sub(ctx1, "sub-1", buf1, []*Channel{channel}, false)
217		-	}()
218		-	go func() {
219		-	defer wg.Done()
220		-	_ = rr.Sub(ctx2, "sub-2", buf2, []*Channel{channel}, false)
221		-	}()
222		-
223		-	time.Sleep(200 * time.Millisecond)
224		-
225		-	// Publish some messages with 2 subscribers
226		-	for i := 0; i < 2; i++ {
227		-	wg.Add(1)
228		-	idx := i
229		-	go func() {
230		-	defer wg.Done()
231		-	msg := fmt.Sprintf("msg%d\n", idx)
232		-	_ = rr.Pub(context.TODO(), "pub", &Buffer{b: bytes.NewBufferString(msg)}, []Channel{channel}, false)
233		-	}()
234		-	}
235		-	time.Sleep(100 * time.Millisecond)
236		-
237		-	// Remove sub-1
238		-	cancel1()
239		-	time.Sleep(200 * time.Millisecond)
240		-
241		-	// Add sub-3
242		-	ctx3, cancel3 := context.WithCancel(context.Background())
243		-	wg.Add(1)
244		-	go func() {
245		-	defer wg.Done()
246		-	_ = rr.Sub(ctx3, "sub-3", buf3, []*Channel{channel}, false)
247		-	}()
248		-
249		-	time.Sleep(200 * time.Millisecond)
250		-
251		-	// Publish more messages with different subscriber set
252		-	for i := 2; i < 4; i++ {
253		-	wg.Add(1)
254		-	idx := i
255		-	go func() {
256		-	defer wg.Done()
257		-	msg := fmt.Sprintf("msg%d\n", idx)
258		-	_ = rr.Pub(context.TODO(), "pub", &Buffer{b: bytes.NewBufferString(msg)}, []Channel{channel}, false)
259		-	}()
260		-	}
261		-
262		-	wg.Wait()
263		-	cancel2()
264		-	cancel3()
265		-
266		-	t.Logf("sub-1: %d bytes", len(buf1.String()))
267		-	t.Logf("sub-2: %d bytes", len(buf2.String()))
268		-	t.Logf("sub-3: %d bytes", len(buf3.String()))
269		-
270		-	// Verify that messages were delivered (exact distribution depends on timing)
271		-	totalLen := len(buf1.String()) + len(buf2.String()) + len(buf3.String())
272		-	if totalLen == 0 {
273		-	t.Fatal("No messages were delivered after subscriber changes")
274		-	}
275		-	}
276		-
277		-	func TestRoundRobinMultipleChannels(t *testing.T) {
278		-	// Test that each channel maintains independent round-robin state
279		-	rr := NewRoundRobin(slog.Default())
280		-
281		-	ch1 := NewChannel("topic-1")
282		-	ch2 := NewChannel("topic-2")
283		-
284		-	buf1ch1 := new(Buffer)
285		-	buf2ch1 := new(Buffer)
286		-	buf1ch2 := new(Buffer)
287		-	buf2ch2 := new(Buffer)
288		-
289		-	var wg sync.WaitGroup
290		-
291		-	// Subscribe to channel 1
292		-	wg.Add(2)
293		-	go func() {
294		-	defer wg.Done()
295		-	_ = rr.Sub(context.TODO(), "sub-1-ch1", buf1ch1, []*Channel{ch1}, false)
296		-	}()
297		-	go func() {
298		-	defer wg.Done()
299		-	_ = rr.Sub(context.TODO(), "sub-2-ch1", buf2ch1, []*Channel{ch1}, false)
300		-	}()
301		-
302		-	// Subscribe to channel 2
303		-	wg.Add(2)
304		-	go func() {
305		-	defer wg.Done()
306		-	_ = rr.Sub(context.TODO(), "sub-1-ch2", buf1ch2, []*Channel{ch2}, false)
307		-	}()
308		-	go func() {
309		-	defer wg.Done()
310		-	_ = rr.Sub(context.TODO(), "sub-2-ch2", buf2ch2, []*Channel{ch2}, false)
311		-	}()
312		-
313		-	time.Sleep(100 * time.Millisecond)
314		-
315		-	// Publish to channel 1
316		-	wg.Add(2)
317		-	for i := 0; i < 2; i++ {
318		-	idx := i
319		-	go func() {
320		-	defer wg.Done()
321		-	msg := fmt.Sprintf("ch1-msg%d\n", idx)
322		-	_ = rr.Pub(context.TODO(), "pub-1", &Buffer{b: bytes.NewBufferString(msg)}, []Channel{ch1}, false)
323		-	}()
324		-	}
325		-
326		-	// Publish to channel 2
327		-	wg.Add(2)
328		-	for i := 0; i < 2; i++ {
329		-	idx := i
330		-	go func() {
331		-	defer wg.Done()
332		-	msg := fmt.Sprintf("ch2-msg%d\n", idx)
333		-	_ = rr.Pub(context.TODO(), "pub-2", &Buffer{b: bytes.NewBufferString(msg)}, []Channel{ch2}, false)
334		-	}()
335		-	}
336		-
337		-	wg.Wait()
338		-
339		-	t.Logf("ch1-buf1: %s", buf1ch1.String())
340		-	t.Logf("ch1-buf2: %s", buf2ch1.String())
341		-	t.Logf("ch2-buf1: %s", buf1ch2.String())
342		-	t.Logf("ch2-buf2: %s", buf2ch2.String())
343		-
344		-	// Both channels should have distributed messages independently
345		-	ch1Total := len(buf1ch1.String()) + len(buf2ch1.String())
346		-	ch2Total := len(buf1ch2.String()) + len(buf2ch2.String())
347		-
348		-	if ch1Total == 0 {
349		-	t.Fatal("Channel 1 should have received messages")
350		-	}
351		-	if ch2Total == 0 {
352		-	t.Fatal("Channel 2 should have received messages")
353		-	}
354		-	}

pico

Patchset 105.3 on 2026-01-29T01:03:32Z · commit 9a6d19e

Patchset `105.3` on 2026-01-29T01:03:32Z · commit 9a6d19e