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.4` on 2026-02-01T17:16:43Z · commit d0dfc85

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

refactor: fixes

Eric Bower 2026-01-29T02:44:01Z

→ chore: SetDispatch on Broker

Eric Bower 2026-02-01T16:37:59Z

chore: SetDispatch on Broker

Eric Bower 2026-02-01T16:37:59Z

Semantic diff summary

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

pkg/apps/pipe/cli.go

function_declaration pub modified

pkg/pubsub/broker.go

chunk lines 5-11 modified
chunk lines 22-28 modified
method_declaration ensureChannel modified
method_declaration SetDispatcher added

pkg/pubsub/dispatcher.go

type_declaration MulticastDispatcher added
method_declaration Dispatch added
type_declaration RoundRobinDispatcher added

pkg/pubsub/multicast.go

chunk lines 6-11 modified
function_declaration GetSubs modified
type_declaration MulticastDispatcher removed
method_declaration Dispatch removed

pkg/pubsub/roundrobin.go

method_declaration Dispatch removed
type_declaration RoundRobinDispatcher removed

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

......@@ -803,7 +803,7 @@ func (handler *CliHandler) pub(cmd *CliCmd, topic string, clientID string) error
		bk = &psub.RoundRobinDispatcher{}
	}
	channel := psub.NewChannel(name)
	channel.Dispatcher = bk
	_ = handler.PubSub.SetDispatcher(bk, []*psub.Channel{channel})

	err := handler.PubSub.Pub(
		cmd.pipeCtx,

+12 -5 pkg/pubsub/broker.go #

......@@ -5,6 +5,7 @@ import (
	"io"
	"iter"
	"log/slog"
	"reflect"
	"sync"
	"time"

......@@ -21,6 +22,7 @@ type Broker interface {
	GetChannels() iter.Seq2[string, *Channel]
	GetClients() iter.Seq2[string, *Client]
	Connect(*Client, []*Channel) (error, error)
	SetDispatcher(dispatcher MessageDispatcher, channels []*Channel) error
}

type BaseBroker struct {

......@@ -197,13 +199,18 @@ func (b *BaseBroker) Connect(client *Client, channels []*Channel) (error, error)

func (b *BaseBroker) ensureChannel(channel *Channel) *Channel {
	dataChannel, _ := b.Channels.LoadOrStore(channel.Topic, channel)
	// Allow overwriting the dispatcher
	if channel.Dispatcher != nil && dataChannel.Dispatcher == nil {
		dataChannel.Dispatcher = channel.Dispatcher
	}

	dataChannel.Handle()
	return dataChannel
}

func (b *BaseBroker) SetDispatcher(dispatcher MessageDispatcher, channels []*Channel) error {
	for _, channel := range channels {
		dataChannel := b.ensureChannel(channel)
		if reflect.TypeOf(dataChannel.Dispatcher) != reflect.TypeOf(dispatcher) {
			dataChannel.Dispatcher = dispatcher
		}
	}
	return nil
}

var _ Broker = (*BaseBroker)(nil)

+70 -0 pkg/pubsub/dispatcher.go #

......@@ -1,8 +1,78 @@
package pubsub

import (
	"slices"
	"strings"
	"sync"
)

// MessageDispatcher defines how messages are dispatched to subscribers.
type MessageDispatcher interface {
	// Dispatch sends a message to the appropriate subscriber(s).
	// It receives the message, all subscribers, and the channel's sync primitives.
	Dispatch(msg ChannelMessage, subscribers []*Client, channelDone chan struct{}) error
}

// MulticastDispatcher sends each message to all eligible subscribers.
type MulticastDispatcher struct{}

func (d *MulticastDispatcher) Dispatch(msg ChannelMessage, subscribers []*Client, channelDone chan struct{}) error {
	var wg sync.WaitGroup
	for _, client := range subscribers {
		wg.Add(1)
		go func(cl *Client) {
			defer wg.Done()
			select {
			case cl.Data <- msg:
			case <-cl.Done:
			case <-channelDone:
			}
		}(client)
	}
	wg.Wait()
	return nil
}

/*
RoundRobin is a load-balancing broker that distributes published messages
to subscribers using a round-robin algorithm.

Unlike Multicast which sends each message to all subscribers, RoundRobin
sends each message to exactly one subscriber, rotating through the available
subscribers for each published message. This provides load balancing for
message processing.

It maintains independent round-robin state per channel/topic.
*/
type RoundRobinDispatcher struct {
	index uint32
	mu    sync.Mutex
}

func (d *RoundRobinDispatcher) Dispatch(msg ChannelMessage, subscribers []*Client, channelDone chan struct{}) error {
	// If no subscribers, nothing to dispatch
	// BlockWrite behavior at publish time ensures subscribers are present when needed
	if len(subscribers) == 0 {
		return nil
	}

	slices.SortFunc(subscribers, func(a, b *Client) int {
		return strings.Compare(a.ID, b.ID)
	})

	// Select the next subscriber in round-robin order
	d.mu.Lock()
	selectedIdx := int(d.index % uint32(len(subscribers)))
	d.index++
	d.mu.Unlock()

	selectedClient := subscribers[selectedIdx]

	select {
	case selectedClient.Data <- msg:
	case <-selectedClient.Done:
	case <-channelDone:
	}

	return nil
}

+4 -23 pkg/pubsub/multicast.go #

......@@ -6,7 +6,6 @@ import (
	"io"
	"iter"
	"log/slog"
	"sync"

	"github.com/antoniomika/syncmap"
)

......@@ -62,9 +61,11 @@ func (p *Multicast) GetSubs() iter.Seq2[string, *Client] {
func (p *Multicast) connect(ctx context.Context, ID string, rw io.ReadWriter, channels []*Channel, direction ChannelDirection, blockWrite bool, replay, keepAlive bool, dispatcher MessageDispatcher) (error, error) {
	client := NewClient(ID, rw, direction, blockWrite, replay, keepAlive)

	// Set dispatcher on all channels
	// Set dispatcher on all channels (only if not already set)
	for _, ch := range channels {
		ch.Dispatcher = dispatcher
		if ch.Dispatcher == nil {
			ch.Dispatcher = dispatcher
		}
	}

	go func() {

......@@ -87,24 +88,4 @@ func (p *Multicast) Sub(ctx context.Context, ID string, rw io.ReadWriter, channe
	return errors.Join(p.connect(ctx, ID, rw, channels, ChannelDirectionOutput, false, false, keepAlive, &MulticastDispatcher{}))
}

// MulticastDispatcher sends each message to all eligible subscribers.
type MulticastDispatcher struct{}

func (d *MulticastDispatcher) Dispatch(msg ChannelMessage, subscribers []*Client, channelDone chan struct{}) error {
	var wg sync.WaitGroup
	for _, client := range subscribers {
		wg.Add(1)
		go func(cl *Client) {
			defer wg.Done()
			select {
			case cl.Data <- msg:
			case <-cl.Done:
			case <-channelDone:
			}
		}(client)
	}
	wg.Wait()
	return nil
}

var _ PubSub = (*Multicast)(nil)

+0 -51 pkg/pubsub/roundrobin.go #

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

import (
	"slices"
	"strings"
	"sync"
)

/*
RoundRobin is a load-balancing broker that distributes published messages
to subscribers using a round-robin algorithm.

Unlike Multicast which sends each message to all subscribers, RoundRobin
sends each message to exactly one subscriber, rotating through the available
subscribers for each published message. This provides load balancing for
message processing.

It maintains independent round-robin state per channel/topic.
*/
type RoundRobinDispatcher struct {
	index uint32
	mu    sync.Mutex
}

func (d *RoundRobinDispatcher) Dispatch(msg ChannelMessage, subscribers []*Client, channelDone chan struct{}) error {
	// If no subscribers, nothing to dispatch
	// BlockWrite behavior at publish time ensures subscribers are present when needed
	if len(subscribers) == 0 {
		return nil
	}

	slices.SortFunc(subscribers, func(a, b *Client) int {
		return strings.Compare(a.ID, b.ID)
	})

	// Select the next subscriber in round-robin order
	d.mu.Lock()
	selectedIdx := int(d.index % uint32(len(subscribers)))
	d.index++
	d.mu.Unlock()

	selectedClient := subscribers[selectedIdx]

	select {
	case selectedClient.Data <- msg:
	case <-selectedClient.Done:
	case <-channelDone:
	}

	return nil
}

...	...		@@ -803,7 +803,7 @@ func (handler CliHandler) pub(cmd CliCmd, topic string, clientID string) error
803	803		bk = &psub.RoundRobinDispatcher{}
804	804		}
805	805		channel := psub.NewChannel(name)
806		-	channel.Dispatcher = bk
	806	+	_ = handler.PubSub.SetDispatcher(bk, []*psub.Channel{channel})
807	807
808	808		err := handler.PubSub.Pub(
809	809		cmd.pipeCtx,

...	...		@@ -197,13 +199,18 @@ func (b BaseBroker) Connect(client Client, channels []*Channel) (error, error)
197	199
198	200		func (b BaseBroker) ensureChannel(channel Channel) *Channel {
199	201		dataChannel, _ := b.Channels.LoadOrStore(channel.Topic, channel)
200		-	// Allow overwriting the dispatcher
201		-	if channel.Dispatcher != nil && dataChannel.Dispatcher == nil {
202		-	dataChannel.Dispatcher = channel.Dispatcher
203		-	}
204		-
205	202		dataChannel.Handle()
206	203		return dataChannel
207	204		}
208	205
	206	+	func (b BaseBroker) SetDispatcher(dispatcher MessageDispatcher, channels []Channel) error {
	207	+	for _, channel := range channels {
	208	+	dataChannel := b.ensureChannel(channel)
	209	+	if reflect.TypeOf(dataChannel.Dispatcher) != reflect.TypeOf(dispatcher) {
	210	+	dataChannel.Dispatcher = dispatcher
	211	+	}
	212	+	}
	213	+	return nil
	214	+	}
	215	+
209	216		var _ Broker = (*BaseBroker)(nil)

...	...		@@ -5,6 +5,7 @@ import (
5	5		"io"
6	6		"iter"
7	7		"log/slog"
	8	+	"reflect"
8	9		"sync"
9	10		"time"
10	11

...	...		@@ -21,6 +22,7 @@ type Broker interface {
21	22		GetChannels() iter.Seq2[string, *Channel]
22	23		GetClients() iter.Seq2[string, *Client]
23	24		Connect(Client, []Channel) (error, error)
	25	+	SetDispatcher(dispatcher MessageDispatcher, channels []*Channel) error
24	26		}
25	27
26	28		type BaseBroker struct {

...	...		@@ -1,8 +1,78 @@
1	1		package pubsub
2	2
	3	+	import (
	4	+	"slices"
	5	+	"strings"
	6	+	"sync"
	7	+	)
	8	+
3	9		// MessageDispatcher defines how messages are dispatched to subscribers.
4	10		type MessageDispatcher interface {
5	11		// Dispatch sends a message to the appropriate subscriber(s).
6	12		// It receives the message, all subscribers, and the channel's sync primitives.
7	13		Dispatch(msg ChannelMessage, subscribers []*Client, channelDone chan struct{}) error
8	14		}
	15	+
	16	+	// MulticastDispatcher sends each message to all eligible subscribers.
	17	+	type MulticastDispatcher struct{}
	18	+
	19	+	func (d MulticastDispatcher) Dispatch(msg ChannelMessage, subscribers []Client, channelDone chan struct{}) error {
	20	+	var wg sync.WaitGroup
	21	+	for _, client := range subscribers {
	22	+	wg.Add(1)
	23	+	go func(cl *Client) {
	24	+	defer wg.Done()
	25	+	select {
	26	+	case cl.Data <- msg:
	27	+	case <-cl.Done:
	28	+	case <-channelDone:
	29	+	}
	30	+	}(client)
	31	+	}
	32	+	wg.Wait()
	33	+	return nil
	34	+	}
	35	+
	36	+	/*
	37	+	RoundRobin is a load-balancing broker that distributes published messages
	38	+	to subscribers using a round-robin algorithm.
	39	+
	40	+	Unlike Multicast which sends each message to all subscribers, RoundRobin
	41	+	sends each message to exactly one subscriber, rotating through the available
	42	+	subscribers for each published message. This provides load balancing for
	43	+	message processing.
	44	+
	45	+	It maintains independent round-robin state per channel/topic.
	46	+	*/
	47	+	type RoundRobinDispatcher struct {
	48	+	index uint32
	49	+	mu sync.Mutex
	50	+	}
	51	+
	52	+	func (d RoundRobinDispatcher) Dispatch(msg ChannelMessage, subscribers []Client, channelDone chan struct{}) error {
	53	+	// If no subscribers, nothing to dispatch
	54	+	// BlockWrite behavior at publish time ensures subscribers are present when needed
	55	+	if len(subscribers) == 0 {
	56	+	return nil
	57	+	}
	58	+
	59	+	slices.SortFunc(subscribers, func(a, b *Client) int {
	60	+	return strings.Compare(a.ID, b.ID)
	61	+	})
	62	+
	63	+	// Select the next subscriber in round-robin order
	64	+	d.mu.Lock()
	65	+	selectedIdx := int(d.index % uint32(len(subscribers)))
	66	+	d.index++
	67	+	d.mu.Unlock()
	68	+
	69	+	selectedClient := subscribers[selectedIdx]
	70	+
	71	+	select {
	72	+	case selectedClient.Data <- msg:
	73	+	case <-selectedClient.Done:
	74	+	case <-channelDone:
	75	+	}
	76	+
	77	+	return nil
	78	+	}

...	...		@@ -6,7 +6,6 @@ import (
6	6		"io"
7	7		"iter"
8	8		"log/slog"
9		-	"sync"
10	9
11	10		"github.com/antoniomika/syncmap"
12	11		)

...	...		@@ -62,9 +61,11 @@ func (p Multicast) GetSubs() iter.Seq2[string, Client] {
62	61		func (p Multicast) connect(ctx context.Context, ID string, rw io.ReadWriter, channels []Channel, direction ChannelDirection, blockWrite bool, replay, keepAlive bool, dispatcher MessageDispatcher) (error, error) {
63	62		client := NewClient(ID, rw, direction, blockWrite, replay, keepAlive)
64	63
65		-	// Set dispatcher on all channels
	64	+	// Set dispatcher on all channels (only if not already set)
66	65		for _, ch := range channels {
67		-	ch.Dispatcher = dispatcher
	66	+	if ch.Dispatcher == nil {
	67	+	ch.Dispatcher = dispatcher
	68	+	}
68	69		}
69	70
70	71		go func() {

...	...		@@ -87,24 +88,4 @@ func (p *Multicast) Sub(ctx context.Context, ID string, rw io.ReadWriter, channe
87	88		return errors.Join(p.connect(ctx, ID, rw, channels, ChannelDirectionOutput, false, false, keepAlive, &MulticastDispatcher{}))
88	89		}
89	90
90		-	// MulticastDispatcher sends each message to all eligible subscribers.
91		-	type MulticastDispatcher struct{}
92		-
93		-	func (d MulticastDispatcher) Dispatch(msg ChannelMessage, subscribers []Client, channelDone chan struct{}) error {
94		-	var wg sync.WaitGroup
95		-	for _, client := range subscribers {
96		-	wg.Add(1)
97		-	go func(cl *Client) {
98		-	defer wg.Done()
99		-	select {
100		-	case cl.Data <- msg:
101		-	case <-cl.Done:
102		-	case <-channelDone:
103		-	}
104		-	}(client)
105		-	}
106		-	wg.Wait()
107		-	return nil
108		-	}
109		-
110	91		var _ PubSub = (*Multicast)(nil)

...	...		@@ -1,51 +0,0 @@
1		-	package pubsub
2		-
3		-	import (
4		-	"slices"
5		-	"strings"
6		-	"sync"
7		-	)
8		-
9		-	/*
10		-	RoundRobin is a load-balancing broker that distributes published messages
11		-	to subscribers using a round-robin algorithm.
12		-
13		-	Unlike Multicast which sends each message to all subscribers, RoundRobin
14		-	sends each message to exactly one subscriber, rotating through the available
15		-	subscribers for each published message. This provides load balancing for
16		-	message processing.
17		-
18		-	It maintains independent round-robin state per channel/topic.
19		-	*/
20		-	type RoundRobinDispatcher struct {
21		-	index uint32
22		-	mu sync.Mutex
23		-	}
24		-
25		-	func (d RoundRobinDispatcher) Dispatch(msg ChannelMessage, subscribers []Client, channelDone chan struct{}) error {
26		-	// If no subscribers, nothing to dispatch
27		-	// BlockWrite behavior at publish time ensures subscribers are present when needed
28		-	if len(subscribers) == 0 {
29		-	return nil
30		-	}
31		-
32		-	slices.SortFunc(subscribers, func(a, b *Client) int {
33		-	return strings.Compare(a.ID, b.ID)
34		-	})
35		-
36		-	// Select the next subscriber in round-robin order
37		-	d.mu.Lock()
38		-	selectedIdx := int(d.index % uint32(len(subscribers)))
39		-	d.index++
40		-	d.mu.Unlock()
41		-
42		-	selectedClient := subscribers[selectedIdx]
43		-
44		-	select {
45		-	case selectedClient.Data <- msg:
46		-	case <-selectedClient.Done:
47		-	case <-channelDone:
48		-	}
49		-
50		-	return nil
51		-	}

pico

Patchset 105.4 on 2026-02-01T17:16:43Z · commit d0dfc85

Patchset `105.4` on 2026-02-01T17:16:43Z · commit d0dfc85