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 4fff471

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

refactor: fixes

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

Semantic diff summary

1 added, 5 modified, 1 signature changed, 1 removed across 4 analyzed files

pkg/pubsub/broker.go

function_declaration Connect modified

pkg/pubsub/channel.go

method_declaration Handle added
function_declaration Handle modified

pkg/pubsub/dispatcher.go

function_declaration dispatcherForGetClients removed

pkg/pubsub/multicast.go

method_declaration connect signature changed
method_declaration Pipe modified
method_declaration Pub modified
method_declaration Sub modified

+0 -14 pkg/pubsub/broker.go #

......@@ -105,20 +105,6 @@ func (b *BaseBroker) Connect(client *Client, channels []*Channel) (error, error)
				data := make([]byte, 32*1024)
				n, err := client.ReadWriter.Read(data)

				// TODO: Skip empty reads
				/*
					if err != nil {
						if errors.Is(err, io.EOF) {
							return
						}
						inputErr = err
						return
					}

					if n == 0 {
						continue
					}
				*/
				data = data[:n]

				channelMessage := ChannelMessage{

+12 -8 pkg/pubsub/channel.go #

......@@ -71,11 +71,6 @@ 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() {

......@@ -100,10 +95,19 @@ func (c *Channel) Handle() {
					}

					// Collect eligible subscribers
					subscribers := dispatcherForGetClients(c.GetClients(), data)
					subscribers := make([]*Client, 0)
					for _, client := range c.GetClients() {
						// Skip input-only clients and senders (unless replay is enabled)
						if client.Direction == ChannelDirectionInput || (client.ID == data.ClientID && !client.Replay) {
							continue
						}
						subscribers = append(subscribers, client)
					}

					// Dispatch message using the configured dispatcher
					_ = c.Dispatcher.Dispatch(data, subscribers, c.Done)
					if len(data.Data) > 0 {
						// Dispatch message using the configured dispatcher
						_ = c.Dispatcher.Dispatch(data, subscribers, c.Done)
					}
				}
			}
		}()

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

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

import (
	"iter"
)

// 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
}

// dispatcherForGetClients collects eligible clients for dispatching.
// Returns clients that should receive messages (output direction, not the sender unless replay).
func dispatcherForGetClients(getClients iter.Seq2[string, *Client], msg ChannelMessage) []*Client {
	subscribers := make([]*Client, 0)
	for _, client := range getClients {
		// Skip input-only clients and senders (unless replay is enabled)
		if client.Direction == ChannelDirectionInput || (client.ID == msg.ClientID && !client.Replay) {
			continue
		}
		subscribers = append(subscribers, client)
	}
	return subscribers
}

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

......@@ -59,9 +59,14 @@ func (p *Multicast) GetSubs() iter.Seq2[string, *Client] {
	return p.getClients(ChannelDirectionOutput)
}

func (p *Multicast) connect(ctx context.Context, ID string, rw io.ReadWriter, channels []*Channel, direction ChannelDirection, blockWrite bool, replay, keepAlive bool) (error, error) {
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
	for _, ch := range channels {
		ch.Dispatcher = dispatcher
	}

	go func() {
		<-ctx.Done()
		client.Cleanup()

......@@ -71,15 +76,15 @@ func (p *Multicast) connect(ctx context.Context, ID string, rw io.ReadWriter, ch
}

func (p *Multicast) Pipe(ctx context.Context, ID string, rw io.ReadWriter, channels []*Channel, replay bool) (error, error) {
	return p.connect(ctx, ID, rw, channels, ChannelDirectionInputOutput, false, replay, false)
	return p.connect(ctx, ID, rw, channels, ChannelDirectionInputOutput, false, replay, false, &MulticastDispatcher{})
}

func (p *Multicast) Pub(ctx context.Context, ID string, rw io.ReadWriter, channels []*Channel, blockWrite bool) error {
	return errors.Join(p.connect(ctx, ID, rw, channels, ChannelDirectionInput, blockWrite, false, false))
	return errors.Join(p.connect(ctx, ID, rw, channels, ChannelDirectionInput, blockWrite, false, false, &MulticastDispatcher{}))
}

func (p *Multicast) Sub(ctx context.Context, ID string, rw io.ReadWriter, channels []*Channel, keepAlive bool) error {
	return errors.Join(p.connect(ctx, ID, rw, channels, ChannelDirectionOutput, false, false, keepAlive))
	return errors.Join(p.connect(ctx, ID, rw, channels, ChannelDirectionOutput, false, false, keepAlive, &MulticastDispatcher{}))
}

// MulticastDispatcher sends each message to all eligible subscribers.

...	...		@@ -105,20 +105,6 @@ func (b BaseBroker) Connect(client Client, channels []*Channel) (error, error)
105	105		data := make([]byte, 32*1024)
106	106		n, err := client.ReadWriter.Read(data)
107	107
108		-	// TODO: Skip empty reads
109		-	/*
110		-	if err != nil {
111		-	if errors.Is(err, io.EOF) {
112		-	return
113		-	}
114		-	inputErr = err
115		-	return
116		-	}
117		-
118		-	if n == 0 {
119		-	continue
120		-	}
121		-	*/
122	108		data = data[:n]
123	109
124	110		channelMessage := ChannelMessage{

...	...		@@ -71,15 +76,15 @@ func (p *Multicast) connect(ctx context.Context, ID string, rw io.ReadWriter, ch
71	76		}
72	77
73	78		func (p Multicast) 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)
	79	+	return p.connect(ctx, ID, rw, channels, ChannelDirectionInputOutput, false, replay, false, &MulticastDispatcher{})
75	80		}
76	81
77	82		func (p Multicast) 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))
	83	+	return errors.Join(p.connect(ctx, ID, rw, channels, ChannelDirectionInput, blockWrite, false, false, &MulticastDispatcher{}))
79	84		}
80	85
81	86		func (p Multicast) 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))
	87	+	return errors.Join(p.connect(ctx, ID, rw, channels, ChannelDirectionOutput, false, false, keepAlive, &MulticastDispatcher{}))
83	88		}
84	89
85	90		// MulticastDispatcher sends each message to all eligible subscribers.

...	...		@@ -71,11 +71,6 @@ 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		-
79	74		c.handleOnce.Do(func() {
80	75		go func() {
81	76		defer func() {

...	...		@@ -100,10 +95,19 @@ func (c *Channel) Handle() {
100	95		}
101	96
102	97		// Collect eligible subscribers
103		-	subscribers := dispatcherForGetClients(c.GetClients(), data)
	98	+	subscribers := make([]*Client, 0)
	99	+	for _, client := range c.GetClients() {
	100	+	// Skip input-only clients and senders (unless replay is enabled)
	101	+	if client.Direction == ChannelDirectionInput \|\| (client.ID == data.ClientID && !client.Replay) {
	102	+	continue
	103	+	}
	104	+	subscribers = append(subscribers, client)
	105	+	}
104	106
105		-	// Dispatch message using the configured dispatcher
106		-	_ = c.Dispatcher.Dispatch(data, subscribers, c.Done)
	107	+	if len(data.Data) > 0 {
	108	+	// Dispatch message using the configured dispatcher
	109	+	_ = c.Dispatcher.Dispatch(data, subscribers, c.Done)
	110	+	}
107	111		}
108	112		}
109	113		}()

...	...		@@ -1,26 +1,8 @@
1	1		package pubsub
2	2
3		-	import (
4		-	"iter"
5		-	)
6		-
7	3		// MessageDispatcher defines how messages are dispatched to subscribers.
8	4		type MessageDispatcher interface {
9	5		// Dispatch sends a message to the appropriate subscriber(s).
10	6		// It receives the message, all subscribers, and the channel's sync primitives.
11	7		Dispatch(msg ChannelMessage, subscribers []*Client, channelDone chan struct{}) error
12	8		}
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		-	}

...	...		@@ -59,9 +59,14 @@ func (p Multicast) GetSubs() iter.Seq2[string, Client] {
59	59		return p.getClients(ChannelDirectionOutput)
60	60		}
61	61
62		-	func (p Multicast) connect(ctx context.Context, ID string, rw io.ReadWriter, channels []Channel, direction ChannelDirection, blockWrite bool, replay, keepAlive bool) (error, error) {
	62	+	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	63		client := NewClient(ID, rw, direction, blockWrite, replay, keepAlive)
64	64
	65	+	// Set dispatcher on all channels
	66	+	for _, ch := range channels {
	67	+	ch.Dispatcher = dispatcher
	68	+	}
	69	+
65	70		go func() {
66	71		<-ctx.Done()
67	72		client.Cleanup()

pico

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

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