pico

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

@@ -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)
+	}
+}