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.1 on 2026-01-25T17:03:08Z · commit 501c042

chore(pubsub): add more tests
Eric Bower 2026-01-25T16:38:25Z
 → feat(pubsub): round robin
Eric Bower 2026-01-25T16:38:25Z
+19 -22 pkg/pubsub/channel.go #
......@@ -57,6 +57,7 @@ type Channel struct {
5757 Clients *syncmap.Map[string, *Client]
5858 handleOnce sync.Once
5959 cleanupOnce sync.Once
60+ Dispatcher MessageDispatcher
6061 }
6162
6263 func (c *Channel) GetClients() iter.Seq2[string, *Client] {
......@@ -70,6 +71,11 @@ func (c *Channel) Cleanup() {
7071 }
7172
7273 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+
7379 c.handleOnce.Do(func() {
7480 go func() {
7581 defer func() {
......@@ -83,30 +89,21 @@ func (c *Channel) Handle() {
8389 case <-c.Done:
8490 return
8591 case data, ok := <-c.Data:
86- var wg sync.WaitGroup
87- for _, client := range c.GetClients() {
88- if client.Direction == ChannelDirectionInput || (client.ID == data.ClientID && !client.Replay) {
89- continue
92+ if !ok {
93+ // Channel is closing, close all client data channels
94+ for _, client := range c.GetClients() {
95+ client.onceData.Do(func() {
96+ close(client.Data)
97+ })
9098 }
91-
92- wg.Add(1)
93- go func() {
94- defer wg.Done()
95- if !ok {
96- client.onceData.Do(func() {
97- close(client.Data)
98- })
99- return
100- }
101-
102- select {
103- case client.Data <- data:
104- case <-client.Done:
105- case <-c.Done:
106- }
107- }()
99+ return
108100 }
109- wg.Wait()
101+
102+ // Collect eligible subscribers
103+ subscribers := dispatcherForGetClients(c.GetClients(), data)
104+
105+ // Dispatch message using the configured dispatcher
106+ _ = c.Dispatcher.Dispatch(data, subscribers, c.Done)
110107 }
111108 }
112109 }()
+26 -0 pkg/pubsub/dispatcher.go #
......@@ -0,0 +1,26 @@
1+package pubsub
2+
3+import (
4+ "iter"
5+)
6+
7+// MessageDispatcher defines how messages are dispatched to subscribers.
8+type MessageDispatcher interface {
9+ // Dispatch sends a message to the appropriate subscriber(s).
10+ // It receives the message, all subscribers, and the channel's sync primitives.
11+ Dispatch(msg ChannelMessage, subscribers []*Client, channelDone chan struct{}) error
12+}
13+
14+// dispatcherForGetClients collects eligible clients for dispatching.
15+// Returns clients that should receive messages (output direction, not the sender unless replay).
16+func dispatcherForGetClients(getClients iter.Seq2[string, *Client], msg ChannelMessage) []*Client {
17+ subscribers := make([]*Client, 0)
18+ for _, client := range getClients {
19+ // Skip input-only clients and senders (unless replay is enabled)
20+ if client.Direction == ChannelDirectionInput || (client.ID == msg.ClientID && !client.Replay) {
21+ continue
22+ }
23+ subscribers = append(subscribers, client)
24+ }
25+ return subscribers
26+}
+21 -0 pkg/pubsub/multicast.go #
......@@ -6,6 +6,7 @@ import (
66 "io"
77 "iter"
88 "log/slog"
9+ "sync"
910
1011 "github.com/antoniomika/syncmap"
1112 )
......@@ -81,4 +82,24 @@ func (p *Multicast) Sub(ctx context.Context, ID string, rw io.ReadWriter, channe
8182 return errors.Join(p.connect(ctx, ID, rw, channels, ChannelDirectionOutput, false, false, keepAlive))
8283 }
8384
85+// MulticastDispatcher sends each message to all eligible subscribers.
86+type MulticastDispatcher struct{}
87+
88+func (d *MulticastDispatcher) Dispatch(msg ChannelMessage, subscribers []*Client, channelDone chan struct{}) error {
89+ var wg sync.WaitGroup
90+ for _, client := range subscribers {
91+ wg.Add(1)
92+ go func(cl *Client) {
93+ defer wg.Done()
94+ select {
95+ case cl.Data <- msg:
96+ case <-cl.Done:
97+ case <-channelDone:
98+ }
99+ }(client)
100+ }
101+ wg.Wait()
102+ return nil
103+}
104+
84105 var _ PubSub = (*Multicast)(nil)
+168 -0 pkg/pubsub/roundrobin.go #
......@@ -0,0 +1,168 @@
1+package pubsub
2+
3+import (
4+ "context"
5+ "errors"
6+ "io"
7+ "iter"
8+ "log/slog"
9+ "sync"
10+
11+ "github.com/antoniomika/syncmap"
12+)
13+
14+/*
15+RoundRobin is a load-balancing broker that distributes published messages
16+to subscribers using a round-robin algorithm.
17+
18+Unlike Multicast which sends each message to all subscribers, RoundRobin
19+sends each message to exactly one subscriber, rotating through the available
20+subscribers for each published message. This provides load balancing for
21+message processing.
22+
23+It maintains independent round-robin state per channel/topic.
24+*/
25+type RoundRobin struct {
26+ Broker
27+ Logger *slog.Logger
28+}
29+
30+func NewRoundRobin(logger *slog.Logger) *RoundRobin {
31+ return &RoundRobin{
32+ Logger: logger,
33+ Broker: &BaseBroker{
34+ Channels: syncmap.New[string, *Channel](),
35+ Logger: logger.With(slog.Bool("broker", true)),
36+ },
37+ }
38+}
39+
40+func (p *RoundRobin) getClients(direction ChannelDirection) iter.Seq2[string, *Client] {
41+ return func(yield func(string, *Client) bool) {
42+ for clientID, client := range p.GetClients() {
43+ if client.Direction == direction {
44+ yield(clientID, client)
45+ }
46+ }
47+ }
48+}
49+
50+func (p *RoundRobin) GetPipes() iter.Seq2[string, *Client] {
51+ return p.getClients(ChannelDirectionInputOutput)
52+}
53+
54+func (p *RoundRobin) GetPubs() iter.Seq2[string, *Client] {
55+ return p.getClients(ChannelDirectionInput)
56+}
57+
58+func (p *RoundRobin) GetSubs() iter.Seq2[string, *Client] {
59+ return p.getClients(ChannelDirectionOutput)
60+}
61+
62+func (p *RoundRobin) connect(ctx context.Context, ID string, rw io.ReadWriter, channels []*Channel, direction ChannelDirection, blockWrite bool, replay, keepAlive bool) (error, error) {
63+ client := NewClient(ID, rw, direction, blockWrite, replay, keepAlive)
64+
65+ go func() {
66+ <-ctx.Done()
67+ client.Cleanup()
68+ }()
69+
70+ return p.Connect(client, channels)
71+}
72+
73+func (p *RoundRobin) Pipe(ctx context.Context, ID string, rw io.ReadWriter, channels []*Channel, replay bool) (error, error) {
74+ return p.connect(ctx, ID, rw, channels, ChannelDirectionInputOutput, false, replay, false)
75+}
76+
77+func (p *RoundRobin) Pub(ctx context.Context, ID string, rw io.ReadWriter, channels []*Channel, blockWrite bool) error {
78+ return errors.Join(p.connect(ctx, ID, rw, channels, ChannelDirectionInput, blockWrite, false, false))
79+}
80+
81+func (p *RoundRobin) Sub(ctx context.Context, ID string, rw io.ReadWriter, channels []*Channel, keepAlive bool) error {
82+ return errors.Join(p.connect(ctx, ID, rw, channels, ChannelDirectionOutput, false, false, keepAlive))
83+}
84+
85+// ensureChannel wraps BaseBroker.ensureChannel to set up round-robin dispatcher.
86+func (p *RoundRobin) ensureChannel(channel *Channel) *Channel {
87+ baseBroker := p.Broker.(*BaseBroker)
88+ dataChannel, _ := baseBroker.Channels.LoadOrStore(channel.Topic, channel)
89+ // Set the round-robin dispatcher on the channel
90+ if dataChannel.Dispatcher == nil {
91+ dataChannel.Dispatcher = &RoundRobinDispatcher{}
92+ }
93+ dataChannel.Handle()
94+ return dataChannel
95+}
96+
97+// Override Connect to use our custom ensureChannel.
98+func (p *RoundRobin) Connect(client *Client, channels []*Channel) (error, error) {
99+ for _, channel := range channels {
100+ dataChannel := p.ensureChannel(channel)
101+ dataChannel.Clients.Store(client.ID, client)
102+ client.Channels.Store(dataChannel.Topic, dataChannel)
103+ defer func() {
104+ client.Channels.Delete(channel.Topic)
105+ dataChannel.Clients.Delete(client.ID)
106+
107+ client.Cleanup()
108+
109+ count := 0
110+ for _, cl := range dataChannel.GetClients() {
111+ if cl.Direction == ChannelDirectionInput || cl.Direction == ChannelDirectionInputOutput {
112+ count++
113+ }
114+ }
115+
116+ if count == 0 {
117+ for _, cl := range dataChannel.GetClients() {
118+ if !cl.KeepAlive {
119+ cl.Cleanup()
120+ }
121+ }
122+ }
123+
124+ p.Cleanup()
125+ }()
126+ }
127+
128+ baseBroker := p.Broker.(*BaseBroker)
129+ return baseBroker.Connect(client, channels)
130+}
131+
132+// Cleanup delegates to BaseBroker.
133+func (p *RoundRobin) Cleanup() {
134+ baseBroker := p.Broker.(*BaseBroker)
135+ baseBroker.Cleanup()
136+}
137+
138+// RoundRobinDispatcher sends each message to a single subscriber in round-robin order.
139+type RoundRobinDispatcher struct {
140+ index uint32
141+ mu sync.Mutex
142+}
143+
144+func (d *RoundRobinDispatcher) Dispatch(msg ChannelMessage, subscribers []*Client, channelDone chan struct{}) error {
145+ // If no subscribers, nothing to dispatch
146+ // BlockWrite behavior at publish time ensures subscribers are present when needed
147+ if len(subscribers) == 0 {
148+ return nil
149+ }
150+
151+ // Select the next subscriber in round-robin order
152+ d.mu.Lock()
153+ selectedIdx := int(d.index % uint32(len(subscribers)))
154+ d.index++
155+ d.mu.Unlock()
156+
157+ selectedClient := subscribers[selectedIdx]
158+
159+ select {
160+ case selectedClient.Data <- msg:
161+ case <-selectedClient.Done:
162+ case <-channelDone:
163+ }
164+
165+ return nil
166+}
167+
168+var _ PubSub = (*RoundRobin)(nil)
+354 -0 pkg/pubsub/roundrobin_test.go #
......@@ -0,0 +1,354 @@
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+}
Back to top