Commit 0002674

feat(pipe): subscribe to wildcard topics

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*"

  1@@ -5,13 +5,32 @@ import (
  2 	"io"
  3 	"iter"
  4 	"log/slog"
  5+	"path"
  6 	"reflect"
  7+	"strings"
  8 	"sync"
  9 	"time"
 10 
 11 	"github.com/antoniomika/syncmap"
 12 )
 13 
 14+// HasWildcard checks if a topic string contains the wildcard character (*).
 15+func HasWildcard(topic string) bool {
 16+	return strings.Contains(topic, "*")
 17+}
 18+
 19+// MatchTopic returns true if pattern matches topic exactly or via path.Match wildcarding.
 20+func MatchTopic(pattern, topic string) bool {
 21+	if pattern == topic {
 22+		return true
 23+	}
 24+	if HasWildcard(pattern) {
 25+		matched, err := path.Match(pattern, topic)
 26+		return err == nil && matched
 27+	}
 28+	return false
 29+}
 30+
 31 /*
 32 Broker receives published messages and dispatches the message to the
 33 subscribing clients. An message contains a message topic that clients
 34@@ -67,7 +86,23 @@ func (b *BaseBroker) Connect(client *Client, channels []*Channel) (error, error)
 35 		dataChannel := b.ensureChannel(channel)
 36 		dataChannel.Clients.Store(client.ID, client)
 37 		client.Channels.Store(dataChannel.Topic, dataChannel)
 38+
 39+		// If client is a subscriber and channel.Topic is a wildcard pattern,
 40+		// attach client to all existing concrete channels matching the pattern.
 41+		if (client.Direction == ChannelDirectionOutput || client.Direction == ChannelDirectionInputOutput) && HasWildcard(channel.Topic) {
 42+			for _, existingChannel := range b.GetChannels() {
 43+				if existingChannel.Topic != channel.Topic && !HasWildcard(existingChannel.Topic) && MatchTopic(channel.Topic, existingChannel.Topic) {
 44+					existingChannel.Clients.Store(client.ID, client)
 45+					client.Channels.Store(existingChannel.Topic, existingChannel)
 46+				}
 47+			}
 48+		}
 49+
 50 		defer func() {
 51+			for _, ch := range client.GetChannels() {
 52+				ch.Clients.Delete(client.ID)
 53+				client.Channels.Delete(ch.Topic)
 54+			}
 55 			client.Channels.Delete(channel.Topic)
 56 			dataChannel.Clients.Delete(client.ID)
 57 
 58@@ -83,7 +118,15 @@ func (b *BaseBroker) Connect(client *Client, channels []*Channel) (error, error)
 59 			if count == 0 {
 60 				for _, cl := range dataChannel.GetClients() {
 61 					if !cl.KeepAlive {
 62-						cl.Cleanup()
 63+						otherChannels := 0
 64+						for _, ch := range cl.GetChannels() {
 65+							if ch.Topic != dataChannel.Topic {
 66+								otherChannels++
 67+							}
 68+						}
 69+						if otherChannels == 0 {
 70+							cl.Cleanup()
 71+						}
 72 					}
 73 				}
 74 			}
 75@@ -198,8 +241,24 @@ func (b *BaseBroker) Connect(client *Client, channels []*Channel) (error, error)
 76 }
 77 
 78 func (b *BaseBroker) ensureChannel(channel *Channel) *Channel {
 79-	dataChannel, _ := b.Channels.LoadOrStore(channel.Topic, channel)
 80+	dataChannel, loaded := b.Channels.LoadOrStore(channel.Topic, channel)
 81 	dataChannel.Handle()
 82+
 83+	// If this is a concrete (non-wildcard) channel created for the first time,
 84+	// attach any active wildcard subscribers whose pattern matches dataChannel.Topic.
 85+	if !loaded && !HasWildcard(channel.Topic) {
 86+		for _, existingChannel := range b.GetChannels() {
 87+			if HasWildcard(existingChannel.Topic) && MatchTopic(existingChannel.Topic, channel.Topic) {
 88+				for _, client := range existingChannel.GetClients() {
 89+					if client.Direction == ChannelDirectionOutput || client.Direction == ChannelDirectionInputOutput {
 90+						dataChannel.Clients.Store(client.ID, client)
 91+						client.Channels.Store(dataChannel.Topic, dataChannel)
 92+					}
 93+				}
 94+			}
 95+		}
 96+	}
 97+
 98 	return dataChannel
 99 }
100 

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