pico

feat(pipe): subscribe to wildcard topics rss

created pr with 131.1 on 2026-08-08T14:14:10Z · by c8ef7d19

added 131.2 on 2026-08-08T14:20:26Z · by c8ef7d19


                      1:  a22c024 ! 1:  0002674 feat(pipe): subscribe to wildcard topics

added 131.3 on 2026-08-08T14:27:35Z · by c8ef7d19


                      1:  0002674 = 1:  0002674 feat(pipe): subscribe to wildcard topics


                      -:  ------- > 2:  d444e49 chore: add more tests and ensure block and keepalive work with wildcards

cmds

checkout latest patchset:

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

checkout any patchset in a patch request:

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

add changes to patch request:

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

Patchset `131.3` on 2026-08-08T14:27:35Z · commit 0002674

→ feat(pipe): subscribe to wildcard topics

Eric Bower 2026-08-08T14:09:38Z

chore: add more tests and ensure block and keepalive work with wildcards

Eric Bower 2026-08-08T14:22:41Z

feat(pipe): subscribe to wildcard topics

Eric Bower 2026-08-08T14:09:38Z

Our pipe service now supports wildcard topics `ssh pipe sub metric-drain*`.

This allows users to publish to multiple topics and have them drain into a
single subscriber.

ssh pipe pub metric-drain-x
ssh pipe pub metric-drain-y
ssh pipe sub "metric-drain*"

Semantic diff summary

3 added, 2 modified, 0 signature changed, 0 removed across 2 analyzed files

pkg/pubsub/broker.go

function_declaration HasWildcard added
function_declaration MatchTopic added
function_declaration Connect modified
method_declaration ensureChannel modified

pkg/pubsub/wildcard_test.go

function_declaration TestWildcardSubExistingAndNewTopics added

+61 -2 pkg/pubsub/broker.go #

......@@ -5,13 +5,32 @@ import (
	"io"
	"iter"
	"log/slog"
	"path"
	"reflect"
	"strings"
	"sync"
	"time"

	"github.com/antoniomika/syncmap"
)

// HasWildcard checks if a topic string contains the wildcard character (*).
func HasWildcard(topic string) bool {
	return strings.Contains(topic, "*")
}

// MatchTopic returns true if pattern matches topic exactly or via path.Match wildcarding.
func MatchTopic(pattern, topic string) bool {
	if pattern == topic {
		return true
	}
	if HasWildcard(pattern) {
		matched, err := path.Match(pattern, topic)
		return err == nil && matched
	}
	return false
}

/*
Broker receives published messages and dispatches the message to the
subscribing clients. An message contains a message topic that clients

......@@ -67,7 +86,23 @@ func (b *BaseBroker) Connect(client *Client, channels []*Channel) (error, error)
		dataChannel := b.ensureChannel(channel)
		dataChannel.Clients.Store(client.ID, client)
		client.Channels.Store(dataChannel.Topic, dataChannel)

		// If client is a subscriber and channel.Topic is a wildcard pattern,
		// attach client to all existing concrete channels matching the pattern.
		if (client.Direction == ChannelDirectionOutput || client.Direction == ChannelDirectionInputOutput) && HasWildcard(channel.Topic) {
			for _, existingChannel := range b.GetChannels() {
				if existingChannel.Topic != channel.Topic && !HasWildcard(existingChannel.Topic) && MatchTopic(channel.Topic, existingChannel.Topic) {
					existingChannel.Clients.Store(client.ID, client)
					client.Channels.Store(existingChannel.Topic, existingChannel)
				}
			}
		}

		defer func() {
			for _, ch := range client.GetChannels() {
				ch.Clients.Delete(client.ID)
				client.Channels.Delete(ch.Topic)
			}
			client.Channels.Delete(channel.Topic)
			dataChannel.Clients.Delete(client.ID)

......@@ -83,7 +118,15 @@ func (b *BaseBroker) Connect(client *Client, channels []*Channel) (error, error)
			if count == 0 {
				for _, cl := range dataChannel.GetClients() {
					if !cl.KeepAlive {
						cl.Cleanup()
						otherChannels := 0
						for _, ch := range cl.GetChannels() {
							if ch.Topic != dataChannel.Topic {
								otherChannels++
							}
						}
						if otherChannels == 0 {
							cl.Cleanup()
						}
					}
				}
			}

......@@ -198,8 +241,24 @@ func (b *BaseBroker) Connect(client *Client, channels []*Channel) (error, error)
}

func (b *BaseBroker) ensureChannel(channel *Channel) *Channel {
	dataChannel, _ := b.Channels.LoadOrStore(channel.Topic, channel)
	dataChannel, loaded := b.Channels.LoadOrStore(channel.Topic, channel)
	dataChannel.Handle()

	// If this is a concrete (non-wildcard) channel created for the first time,
	// attach any active wildcard subscribers whose pattern matches dataChannel.Topic.
	if !loaded && !HasWildcard(channel.Topic) {
		for _, existingChannel := range b.GetChannels() {
			if HasWildcard(existingChannel.Topic) && MatchTopic(existingChannel.Topic, channel.Topic) {
				for _, client := range existingChannel.GetClients() {
					if client.Direction == ChannelDirectionOutput || client.Direction == ChannelDirectionInputOutput {
						dataChannel.Clients.Store(client.ID, client)
						client.Channels.Store(dataChannel.Topic, dataChannel)
					}
				}
			}
		}
	}

	return dataChannel
}

+75 -0 pkg/pubsub/wildcard_test.go #

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

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

// TestWildcardSubExistingAndNewTopics verifies that a subscriber with a wildcard topic
// (e.g., "metric-drain*") receives messages published to existing matching sub-topics
// AND any new matching sub-topics created AFTER the subscription was established.
func TestWildcardSubExistingAndNewTopics(t *testing.T) {
	cast := NewMulticast(slog.Default())

	subBuf := new(Buffer)
	subCtx, cancelSub := context.WithCancel(context.Background())
	defer cancelSub()

	// Wildcard subscription topic
	wildcardChannel := NewChannel("metric-drain*")

	var wg sync.WaitGroup

	// Start subscriber listening on wildcard topic "metric-drain*"
	wg.Add(1)
	go func() {
		defer wg.Done()
		_ = cast.Sub(subCtx, "sub-wildcard", subBuf, []*Channel{wildcardChannel}, false)
	}()

	time.Sleep(50 * time.Millisecond)

	// Publish to first topic matching wildcard: "metric-drain-pgs"
	channelPGS := NewChannel("metric-drain-pgs")
	pub1Ctx, cancelPub1 := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancelPub1()

	_ = cast.Pub(pub1Ctx, "pub-pgs", &Buffer{b: *bytes.NewBufferString("pgs-data\n")}, []*Channel{channelPGS}, false)

	// Publish to second topic matching wildcard: "metric-drain-prose"
	channelProse := NewChannel("metric-drain-prose")
	pub2Ctx, cancelPub2 := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancelPub2()

	_ = cast.Pub(pub2Ctx, "pub-prose", &Buffer{b: *bytes.NewBufferString("prose-data\n")}, []*Channel{channelProse}, false)

	// Publish to non-matching topic: "other-topic"
	channelOther := NewChannel("other-topic")
	pub3Ctx, cancelPub3 := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancelPub3()

	_ = cast.Pub(pub3Ctx, "pub-other", &Buffer{b: *bytes.NewBufferString("other-data\n")}, []*Channel{channelOther}, false)

	// Wait briefly for dispatch
	time.Sleep(100 * time.Millisecond)

	// Stop subscriber
	cancelSub()
	wg.Wait()

	got := subBuf.String()

	if !bytes.Contains([]byte(got), []byte("pgs-data\n")) {
		t.Errorf("expected wildcard subscriber to receive pgs-data, got: %q", got)
	}
	if !bytes.Contains([]byte(got), []byte("prose-data\n")) {
		t.Errorf("expected wildcard subscriber to receive prose-data, got: %q", got)
	}
	if bytes.Contains([]byte(got), []byte("other-data\n")) {
		t.Errorf("wildcard subscriber should NOT receive other-data, got: %q", got)
	}
}

...	...		@@ -67,7 +86,23 @@ func (b BaseBroker) Connect(client Client, channels []*Channel) (error, error)
67	86		dataChannel := b.ensureChannel(channel)
68	87		dataChannel.Clients.Store(client.ID, client)
69	88		client.Channels.Store(dataChannel.Topic, dataChannel)
	89	+
	90	+	// If client is a subscriber and channel.Topic is a wildcard pattern,
	91	+	// attach client to all existing concrete channels matching the pattern.
	92	+	if (client.Direction == ChannelDirectionOutput \|\| client.Direction == ChannelDirectionInputOutput) && HasWildcard(channel.Topic) {
	93	+	for _, existingChannel := range b.GetChannels() {
	94	+	if existingChannel.Topic != channel.Topic && !HasWildcard(existingChannel.Topic) && MatchTopic(channel.Topic, existingChannel.Topic) {
	95	+	existingChannel.Clients.Store(client.ID, client)
	96	+	client.Channels.Store(existingChannel.Topic, existingChannel)
	97	+	}
	98	+	}
	99	+	}
	100	+
70	101		defer func() {
	102	+	for _, ch := range client.GetChannels() {
	103	+	ch.Clients.Delete(client.ID)
	104	+	client.Channels.Delete(ch.Topic)
	105	+	}
71	106		client.Channels.Delete(channel.Topic)
72	107		dataChannel.Clients.Delete(client.ID)
73	108

...	...		@@ -83,7 +118,15 @@ func (b BaseBroker) Connect(client Client, channels []*Channel) (error, error)
83	118		if count == 0 {
84	119		for _, cl := range dataChannel.GetClients() {
85	120		if !cl.KeepAlive {
86		-	cl.Cleanup()
	121	+	otherChannels := 0
	122	+	for _, ch := range cl.GetChannels() {
	123	+	if ch.Topic != dataChannel.Topic {
	124	+	otherChannels++
	125	+	}
	126	+	}
	127	+	if otherChannels == 0 {
	128	+	cl.Cleanup()
	129	+	}
87	130		}
88	131		}
89	132		}

...	...		@@ -198,8 +241,24 @@ func (b BaseBroker) Connect(client Client, channels []*Channel) (error, error)
198	241		}
199	242
200	243		func (b BaseBroker) ensureChannel(channel Channel) *Channel {
201		-	dataChannel, _ := b.Channels.LoadOrStore(channel.Topic, channel)
	244	+	dataChannel, loaded := b.Channels.LoadOrStore(channel.Topic, channel)
202	245		dataChannel.Handle()
	246	+
	247	+	// If this is a concrete (non-wildcard) channel created for the first time,
	248	+	// attach any active wildcard subscribers whose pattern matches dataChannel.Topic.
	249	+	if !loaded && !HasWildcard(channel.Topic) {
	250	+	for _, existingChannel := range b.GetChannels() {
	251	+	if HasWildcard(existingChannel.Topic) && MatchTopic(existingChannel.Topic, channel.Topic) {
	252	+	for _, client := range existingChannel.GetClients() {
	253	+	if client.Direction == ChannelDirectionOutput \|\| client.Direction == ChannelDirectionInputOutput {
	254	+	dataChannel.Clients.Store(client.ID, client)
	255	+	client.Channels.Store(dataChannel.Topic, dataChannel)
	256	+	}
	257	+	}
	258	+	}
	259	+	}
	260	+	}
	261	+
203	262		return dataChannel
204	263		}
205	264

...	...		@@ -5,13 +5,32 @@ import (
5	5		"io"
6	6		"iter"
7	7		"log/slog"
	8	+	"path"
8	9		"reflect"
	10	+	"strings"
9	11		"sync"
10	12		"time"
11	13
12	14		"github.com/antoniomika/syncmap"
13	15		)
14	16
	17	+	// HasWildcard checks if a topic string contains the wildcard character (*).
	18	+	func HasWildcard(topic string) bool {
	19	+	return strings.Contains(topic, "*")
	20	+	}
	21	+
	22	+	// MatchTopic returns true if pattern matches topic exactly or via path.Match wildcarding.
	23	+	func MatchTopic(pattern, topic string) bool {
	24	+	if pattern == topic {
	25	+	return true
	26	+	}
	27	+	if HasWildcard(pattern) {
	28	+	matched, err := path.Match(pattern, topic)
	29	+	return err == nil && matched
	30	+	}
	31	+	return false
	32	+	}
	33	+
15	34		/*
16	35		Broker receives published messages and dispatches the message to the
17	36		subscribing clients. An message contains a message topic that clients

...	...		@@ -0,0 +1,75 @@
	1	+	package pubsub
	2	+
	3	+	import (
	4	+	"bytes"
	5	+	"context"
	6	+	"log/slog"
	7	+	"sync"
	8	+	"testing"
	9	+	"time"
	10	+	)
	11	+
	12	+	// TestWildcardSubExistingAndNewTopics verifies that a subscriber with a wildcard topic
	13	+	// (e.g., "metric-drain*") receives messages published to existing matching sub-topics
	14	+	// AND any new matching sub-topics created AFTER the subscription was established.
	15	+	func TestWildcardSubExistingAndNewTopics(t *testing.T) {
	16	+	cast := NewMulticast(slog.Default())
	17	+
	18	+	subBuf := new(Buffer)
	19	+	subCtx, cancelSub := context.WithCancel(context.Background())
	20	+	defer cancelSub()
	21	+
	22	+	// Wildcard subscription topic
	23	+	wildcardChannel := NewChannel("metric-drain*")
	24	+
	25	+	var wg sync.WaitGroup
	26	+
	27	+	// Start subscriber listening on wildcard topic "metric-drain*"
	28	+	wg.Add(1)
	29	+	go func() {
	30	+	defer wg.Done()
	31	+	_ = cast.Sub(subCtx, "sub-wildcard", subBuf, []*Channel{wildcardChannel}, false)
	32	+	}()
	33	+
	34	+	time.Sleep(50 * time.Millisecond)
	35	+
	36	+	// Publish to first topic matching wildcard: "metric-drain-pgs"
	37	+	channelPGS := NewChannel("metric-drain-pgs")
	38	+	pub1Ctx, cancelPub1 := context.WithTimeout(context.Background(), 2*time.Second)
	39	+	defer cancelPub1()
	40	+
	41	+	_ = cast.Pub(pub1Ctx, "pub-pgs", &Buffer{b: bytes.NewBufferString("pgs-data\n")}, []Channel{channelPGS}, false)
	42	+
	43	+	// Publish to second topic matching wildcard: "metric-drain-prose"
	44	+	channelProse := NewChannel("metric-drain-prose")
	45	+	pub2Ctx, cancelPub2 := context.WithTimeout(context.Background(), 2*time.Second)
	46	+	defer cancelPub2()
	47	+
	48	+	_ = cast.Pub(pub2Ctx, "pub-prose", &Buffer{b: bytes.NewBufferString("prose-data\n")}, []Channel{channelProse}, false)
	49	+
	50	+	// Publish to non-matching topic: "other-topic"
	51	+	channelOther := NewChannel("other-topic")
	52	+	pub3Ctx, cancelPub3 := context.WithTimeout(context.Background(), 2*time.Second)
	53	+	defer cancelPub3()
	54	+
	55	+	_ = cast.Pub(pub3Ctx, "pub-other", &Buffer{b: bytes.NewBufferString("other-data\n")}, []Channel{channelOther}, false)
	56	+
	57	+	// Wait briefly for dispatch
	58	+	time.Sleep(100 * time.Millisecond)
	59	+
	60	+	// Stop subscriber
	61	+	cancelSub()
	62	+	wg.Wait()
	63	+
	64	+	got := subBuf.String()
	65	+
	66	+	if !bytes.Contains([]byte(got), []byte("pgs-data\n")) {
	67	+	t.Errorf("expected wildcard subscriber to receive pgs-data, got: %q", got)
	68	+	}
	69	+	if !bytes.Contains([]byte(got), []byte("prose-data\n")) {
	70	+	t.Errorf("expected wildcard subscriber to receive prose-data, got: %q", got)
	71	+	}
	72	+	if bytes.Contains([]byte(got), []byte("other-data\n")) {
	73	+	t.Errorf("wildcard subscriber should NOT receive other-data, got: %q", got)
	74	+	}
	75	+	}

pico

Patchset 131.3 on 2026-08-08T14:27:35Z · commit 0002674

Patchset `131.3` on 2026-08-08T14:27:35Z · commit 0002674