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 d444e49

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
+8 -4 pkg/apps/pipe/cli.go #
......@@ -700,13 +700,15 @@ func (handler *CliHandler) pub(cmd *CliCmd, topic string, clientID string) error
700700 if *block {
701701 count := 0
702702 for topic, channel := range handler.PubSub.GetChannels() {
703- if topic == name {
703+ if topic == name || (psub.HasWildcard(topic) && psub.MatchTopic(topic, name)) {
704704 for _, client := range channel.GetClients() {
705705 if client.Direction == psub.ChannelDirectionOutput || client.Direction == psub.ChannelDirectionInputOutput {
706706 count++
707707 }
708708 }
709- break
709+ if topic == name {
710+ break
711+ }
710712 }
711713 }
712714
......@@ -735,13 +737,15 @@ func (handler *CliHandler) pub(cmd *CliCmd, topic string, clientID string) error
735737 case <-time.After(1 * time.Millisecond):
736738 count := 0
737739 for topic, channel := range handler.PubSub.GetChannels() {
738- if topic == name {
740+ if topic == name || (psub.HasWildcard(topic) && psub.MatchTopic(topic, name)) {
739741 for _, client := range channel.GetClients() {
740742 if client.Direction == psub.ChannelDirectionOutput || client.Direction == psub.ChannelDirectionInputOutput {
741743 count++
742744 }
743745 }
744- break
746+ if topic == name {
747+ break
748+ }
745749 }
746750 }
747751
+106 -0 pkg/apps/pipe/ssh_test.go #
......@@ -1,6 +1,7 @@
11 package pipe
22
33 import (
4+ "bytes"
45 "context"
56 "crypto/ed25519"
67 "crypto/rand"
......@@ -9,6 +10,7 @@ import (
910 "log/slog"
1011 "os"
1112 "strings"
13+ "sync"
1214 "testing"
1315 "time"
1416
......@@ -2200,3 +2202,107 @@ func TestMonitor_FixedWindowNonSliding(t *testing.T) {
22002202 monitor.WindowEnd.Format(time.RFC3339))
22012203 }
22022204 }
2205+
2206+func TestSSH_WildcardSub(t *testing.T) {
2207+ server := NewTestSSHServer(t)
2208+ defer server.Shutdown()
2209+
2210+ user := GenerateUser("alice")
2211+ dbUser := &db.User{ID: "alice-id", Name: "alice"}
2212+ server.DBPool.AddUser(dbUser)
2213+ server.DBPool.AddPubkey(&db.PublicKey{
2214+ ID: "alice-pk",
2215+ UserID: "alice-id",
2216+ Key: user.PublicKey(),
2217+ })
2218+
2219+ client, err := user.NewClient()
2220+ if err != nil {
2221+ t.Fatalf("failed to dial ssh server: %v", err)
2222+ }
2223+ defer func() { _ = client.Close() }()
2224+
2225+ // 1. Subscribe to wildcard topic: "sub metric-drain*"
2226+ subSession, err := client.NewSession()
2227+ if err != nil {
2228+ t.Fatalf("failed to create sub session: %v", err)
2229+ }
2230+
2231+ subOut, err := subSession.StdoutPipe()
2232+ if err != nil {
2233+ t.Fatalf("failed stdout pipe: %v", err)
2234+ }
2235+
2236+ if err := subSession.Start("sub metric-drain*"); err != nil {
2237+ t.Fatalf("failed to start sub: %v", err)
2238+ }
2239+
2240+ var buf bytes.Buffer
2241+ var bufMu sync.Mutex
2242+ go func() {
2243+ b := make([]byte, 1024)
2244+ for {
2245+ n, err := subOut.Read(b)
2246+ if n > 0 {
2247+ bufMu.Lock()
2248+ buf.Write(b[:n])
2249+ bufMu.Unlock()
2250+ }
2251+ if err != nil {
2252+ break
2253+ }
2254+ }
2255+ }()
2256+
2257+ time.Sleep(100 * time.Millisecond)
2258+
2259+ // 2. Publish to "metric-drain-pgs"
2260+ pubClient1, err := user.NewClient()
2261+ if err != nil {
2262+ t.Fatalf("failed to dial pub client 1: %v", err)
2263+ }
2264+ defer func() { _ = pubClient1.Close() }()
2265+
2266+ pubSession1, err := pubClient1.NewSession()
2267+ if err != nil {
2268+ t.Fatalf("failed pub session 1: %v", err)
2269+ }
2270+ pubIn1, _ := pubSession1.StdinPipe()
2271+ go func() {
2272+ defer func() { _ = pubIn1.Close() }()
2273+ _, _ = io.WriteString(pubIn1, "pgs-event\n")
2274+ }()
2275+ _ = pubSession1.Run("pub metric-drain-pgs -b=false")
2276+
2277+ // 3. Publish to "metric-drain-prose"
2278+ pubClient2, err := user.NewClient()
2279+ if err != nil {
2280+ t.Fatalf("failed to dial pub client 2: %v", err)
2281+ }
2282+ defer func() { _ = pubClient2.Close() }()
2283+
2284+ pubSession2, err := pubClient2.NewSession()
2285+ if err != nil {
2286+ t.Fatalf("failed pub session 2: %v", err)
2287+ }
2288+ pubIn2, _ := pubSession2.StdinPipe()
2289+ go func() {
2290+ defer func() { _ = pubIn2.Close() }()
2291+ _, _ = io.WriteString(pubIn2, "prose-event\n")
2292+ }()
2293+ _ = pubSession2.Run("pub metric-drain-prose -b=false")
2294+
2295+ time.Sleep(150 * time.Millisecond)
2296+ _ = subSession.Close()
2297+
2298+ bufMu.Lock()
2299+ output := buf.String()
2300+ bufMu.Unlock()
2301+
2302+ if !strings.Contains(output, "pgs-event") {
2303+ t.Errorf("expected SSH wildcard subscriber output to contain 'pgs-event', got: %q", output)
2304+ }
2305+ if !strings.Contains(output, "prose-event") {
2306+ t.Errorf("expected SSH wildcard subscriber output to contain 'prose-event', got: %q", output)
2307+ }
2308+}
+129 -0 pkg/pubsub/wildcard_test.go #
......@@ -73,3 +73,132 @@ func TestWildcardSubExistingAndNewTopics(t *testing.T) {
7373 t.Errorf("wildcard subscriber should NOT receive other-data, got: %q", got)
7474 }
7575 }
76+
77+// TestWildcardSubMultipleSubscribers verifies that multiple wildcard subscribers
78+// listening on the same pattern both receive published messages.
79+func TestWildcardSubMultipleSubscribers(t *testing.T) {
80+ cast := NewMulticast(slog.Default())
81+
82+ subBuf1 := new(Buffer)
83+ subBuf2 := new(Buffer)
84+ subCtx, cancelSub := context.WithCancel(context.Background())
85+ defer cancelSub()
86+
87+ wildcardChannel := NewChannel("logs-*")
88+
89+ var wg sync.WaitGroup
90+ wg.Add(2)
91+
92+ go func() {
93+ defer wg.Done()
94+ _ = cast.Sub(subCtx, "sub-1", subBuf1, []*Channel{wildcardChannel}, false)
95+ }()
96+
97+ go func() {
98+ defer wg.Done()
99+ _ = cast.Sub(subCtx, "sub-2", subBuf2, []*Channel{wildcardChannel}, false)
100+ }()
101+
102+ time.Sleep(50 * time.Millisecond)
103+
104+ channel := NewChannel("logs-app1")
105+ pubCtx, cancelPub := context.WithTimeout(context.Background(), 2*time.Second)
106+ defer cancelPub()
107+
108+ _ = cast.Pub(pubCtx, "pub-1", &Buffer{b: *bytes.NewBufferString("app1-log\n")}, []*Channel{channel}, false)
109+
110+ time.Sleep(100 * time.Millisecond)
111+ cancelSub()
112+ wg.Wait()
113+
114+ if subBuf1.String() != "app1-log\n" {
115+ t.Errorf("sub-1 expected app1-log, got %q", subBuf1.String())
116+ }
117+ if subBuf2.String() != "app1-log\n" {
118+ t.Errorf("sub-2 expected app1-log, got %q", subBuf2.String())
119+ }
120+}
121+
122+// TestWildcardSubVariousPatterns verifies prefix, suffix, and middle asterisk wildcard matching.
123+func TestWildcardSubVariousPatterns(t *testing.T) {
124+ cast := NewMulticast(slog.Default())
125+
126+ prefixBuf := new(Buffer)
127+ suffixBuf := new(Buffer)
128+ middleBuf := new(Buffer)
129+
130+ ctx, cancel := context.WithCancel(context.Background())
131+ defer cancel()
132+
133+ var wg sync.WaitGroup
134+ wg.Add(3)
135+
136+ go func() {
137+ defer wg.Done()
138+ _ = cast.Sub(ctx, "sub-prefix", prefixBuf, []*Channel{NewChannel("metric-*")}, false)
139+ }()
140+ go func() {
141+ defer wg.Done()
142+ _ = cast.Sub(ctx, "sub-suffix", suffixBuf, []*Channel{NewChannel("*-drain")}, false)
143+ }()
144+ go func() {
145+ defer wg.Done()
146+ _ = cast.Sub(ctx, "sub-middle", middleBuf, []*Channel{NewChannel("metric-*-drain")}, false)
147+ }()
148+
149+ time.Sleep(50 * time.Millisecond)
150+
151+ pubCtx, cancelPub := context.WithTimeout(context.Background(), 2*time.Second)
152+ defer cancelPub()
153+
154+ // Publish to metric-app-drain
155+ _ = cast.Pub(pubCtx, "pub", &Buffer{b: *bytes.NewBufferString("event\n")}, []*Channel{NewChannel("metric-app-drain")}, false)
156+
157+ time.Sleep(100 * time.Millisecond)
158+ cancel()
159+ wg.Wait()
160+
161+ if prefixBuf.String() != "event\n" {
162+ t.Errorf("prefix subscriber expected event, got %q", prefixBuf.String())
163+ }
164+ if suffixBuf.String() != "event\n" {
165+ t.Errorf("suffix subscriber expected event, got %q", suffixBuf.String())
166+ }
167+ if middleBuf.String() != "event\n" {
168+ t.Errorf("middle subscriber expected event, got %q", middleBuf.String())
169+ }
170+}
171+
172+// TestWildcardSubLiteralCharsWithoutStar verifies that '?' or '[' without '*' are treated as literal names.
173+func TestWildcardSubLiteralCharsWithoutStar(t *testing.T) {
174+ cast := NewMulticast(slog.Default())
175+
176+ subBuf := new(Buffer)
177+ ctx, cancel := context.WithCancel(context.Background())
178+ defer cancel()
179+
180+ var wg sync.WaitGroup
181+ wg.Add(1)
182+
183+ // Subscribe to a topic with a literal '?' character
184+ go func() {
185+ defer wg.Done()
186+ _ = cast.Sub(ctx, "sub-literal", subBuf, []*Channel{NewChannel("topic?one")}, false)
187+ }()
188+
189+ time.Sleep(50 * time.Millisecond)
190+
191+ pubCtx, cancelPub := context.WithTimeout(context.Background(), 2*time.Second)
192+ defer cancelPub()
193+
194+ // Publish to topicXone (should NOT match because '?' is not treated as a wildcard)
195+ _ = cast.Pub(pubCtx, "pub", &Buffer{b: *bytes.NewBufferString("data\n")}, []*Channel{NewChannel("topicXone")}, false)
196+
197+ time.Sleep(100 * time.Millisecond)
198+ cancel()
199+ wg.Wait()
200+
201+ if subBuf.String() != "" {
202+ t.Errorf("literal subscriber should not have matched topicXone, got: %q", subBuf.String())
203+ }
204+}
Back to top