pico

feat(pipe): subscribe to wildcard topics rss
created pr with pico/131.1 on 2026-08-08T14:14:10Z · by c8ef7d19
added pico/131.2 on 2026-08-08T14:20:26Z · by c8ef7d19
1: a22c024 ! 1: 0002674 feat(pipe): subscribe to wildcard topics
added pico/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 pull pico:131 | git am -3
checkout specific patchset revision: 
ssh pr.pico.sh pull pico:131 [rev] | git am -3
add changes to patch request: 
git format-patch main --stdout | ssh pr.pico.sh pico:131
add comment to patch request: 
ssh pr.pico.sh comment pico:131 "lgtm!"

Patchset pico/131.1 on 2026-08-08T14:14:10Z · commit a22c024

→ feat(pipe): subscribe to wildcard topics
Eric Bower 2026-08-08T14:09:38Z
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
1 added, 0 modified, 0 signature changed, 0 removed across 1 analyzed file
pkg/pubsub/wildcard_test.go
+75 -0 pkg/pubsub/wildcard_test.go #
......@@ -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+}
Back to top