Commit d444e49

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

 1@@ -700,13 +700,15 @@ func (handler *CliHandler) pub(cmd *CliCmd, topic string, clientID string) error
 2 	if *block {
 3 		count := 0
 4 		for topic, channel := range handler.PubSub.GetChannels() {
 5-			if topic == name {
 6+			if topic == name || (psub.HasWildcard(topic) && psub.MatchTopic(topic, name)) {
 7 				for _, client := range channel.GetClients() {
 8 					if client.Direction == psub.ChannelDirectionOutput || client.Direction == psub.ChannelDirectionInputOutput {
 9 						count++
10 					}
11 				}
12-				break
13+				if topic == name {
14+					break
15+				}
16 			}
17 		}
18 
19@@ -735,13 +737,15 @@ func (handler *CliHandler) pub(cmd *CliCmd, topic string, clientID string) error
20 					case <-time.After(1 * time.Millisecond):
21 						count := 0
22 						for topic, channel := range handler.PubSub.GetChannels() {
23-							if topic == name {
24+							if topic == name || (psub.HasWildcard(topic) && psub.MatchTopic(topic, name)) {
25 								for _, client := range channel.GetClients() {
26 									if client.Direction == psub.ChannelDirectionOutput || client.Direction == psub.ChannelDirectionInputOutput {
27 										count++
28 									}
29 								}
30-								break
31+								if topic == name {
32+									break
33+								}
34 							}
35 						}
36 

  1@@ -1,6 +1,7 @@
  2 package pipe
  3 
  4 import (
  5+	"bytes"
  6 	"context"
  7 	"crypto/ed25519"
  8 	"crypto/rand"
  9@@ -9,6 +10,7 @@ import (
 10 	"log/slog"
 11 	"os"
 12 	"strings"
 13+	"sync"
 14 	"testing"
 15 	"time"
 16 
 17@@ -2200,3 +2202,107 @@ func TestMonitor_FixedWindowNonSliding(t *testing.T) {
 18 			monitor.WindowEnd.Format(time.RFC3339))
 19 	}
 20 }
 21+
 22+func TestSSH_WildcardSub(t *testing.T) {
 23+	server := NewTestSSHServer(t)
 24+	defer server.Shutdown()
 25+
 26+	user := GenerateUser("alice")
 27+	dbUser := &db.User{ID: "alice-id", Name: "alice"}
 28+	server.DBPool.AddUser(dbUser)
 29+	server.DBPool.AddPubkey(&db.PublicKey{
 30+		ID:     "alice-pk",
 31+		UserID: "alice-id",
 32+		Key:    user.PublicKey(),
 33+	})
 34+
 35+	client, err := user.NewClient()
 36+	if err != nil {
 37+		t.Fatalf("failed to dial ssh server: %v", err)
 38+	}
 39+	defer func() { _ = client.Close() }()
 40+
 41+	// 1. Subscribe to wildcard topic: "sub metric-drain*"
 42+	subSession, err := client.NewSession()
 43+	if err != nil {
 44+		t.Fatalf("failed to create sub session: %v", err)
 45+	}
 46+
 47+	subOut, err := subSession.StdoutPipe()
 48+	if err != nil {
 49+		t.Fatalf("failed stdout pipe: %v", err)
 50+	}
 51+
 52+	if err := subSession.Start("sub metric-drain*"); err != nil {
 53+		t.Fatalf("failed to start sub: %v", err)
 54+	}
 55+
 56+	var buf bytes.Buffer
 57+	var bufMu sync.Mutex
 58+	go func() {
 59+		b := make([]byte, 1024)
 60+		for {
 61+			n, err := subOut.Read(b)
 62+			if n > 0 {
 63+				bufMu.Lock()
 64+				buf.Write(b[:n])
 65+				bufMu.Unlock()
 66+			}
 67+			if err != nil {
 68+				break
 69+			}
 70+		}
 71+	}()
 72+
 73+	time.Sleep(100 * time.Millisecond)
 74+
 75+	// 2. Publish to "metric-drain-pgs"
 76+	pubClient1, err := user.NewClient()
 77+	if err != nil {
 78+		t.Fatalf("failed to dial pub client 1: %v", err)
 79+	}
 80+	defer func() { _ = pubClient1.Close() }()
 81+
 82+	pubSession1, err := pubClient1.NewSession()
 83+	if err != nil {
 84+		t.Fatalf("failed pub session 1: %v", err)
 85+	}
 86+	pubIn1, _ := pubSession1.StdinPipe()
 87+	go func() {
 88+		defer func() { _ = pubIn1.Close() }()
 89+		_, _ = io.WriteString(pubIn1, "pgs-event\n")
 90+	}()
 91+	_ = pubSession1.Run("pub metric-drain-pgs -b=false")
 92+
 93+	// 3. Publish to "metric-drain-prose"
 94+	pubClient2, err := user.NewClient()
 95+	if err != nil {
 96+		t.Fatalf("failed to dial pub client 2: %v", err)
 97+	}
 98+	defer func() { _ = pubClient2.Close() }()
 99+
100+	pubSession2, err := pubClient2.NewSession()
101+	if err != nil {
102+		t.Fatalf("failed pub session 2: %v", err)
103+	}
104+	pubIn2, _ := pubSession2.StdinPipe()
105+	go func() {
106+		defer func() { _ = pubIn2.Close() }()
107+		_, _ = io.WriteString(pubIn2, "prose-event\n")
108+	}()
109+	_ = pubSession2.Run("pub metric-drain-prose -b=false")
110+
111+	time.Sleep(150 * time.Millisecond)
112+	_ = subSession.Close()
113+
114+	bufMu.Lock()
115+	output := buf.String()
116+	bufMu.Unlock()
117+
118+	if !strings.Contains(output, "pgs-event") {
119+		t.Errorf("expected SSH wildcard subscriber output to contain 'pgs-event', got: %q", output)
120+	}
121+	if !strings.Contains(output, "prose-event") {
122+		t.Errorf("expected SSH wildcard subscriber output to contain 'prose-event', got: %q", output)
123+	}
124+}

  1@@ -73,3 +73,132 @@ func TestWildcardSubExistingAndNewTopics(t *testing.T) {
  2 		t.Errorf("wildcard subscriber should NOT receive other-data, got: %q", got)
  3 	}
  4 }
  5+
  6+// TestWildcardSubMultipleSubscribers verifies that multiple wildcard subscribers
  7+// listening on the same pattern both receive published messages.
  8+func TestWildcardSubMultipleSubscribers(t *testing.T) {
  9+	cast := NewMulticast(slog.Default())
 10+
 11+	subBuf1 := new(Buffer)
 12+	subBuf2 := new(Buffer)
 13+	subCtx, cancelSub := context.WithCancel(context.Background())
 14+	defer cancelSub()
 15+
 16+	wildcardChannel := NewChannel("logs-*")
 17+
 18+	var wg sync.WaitGroup
 19+	wg.Add(2)
 20+
 21+	go func() {
 22+		defer wg.Done()
 23+		_ = cast.Sub(subCtx, "sub-1", subBuf1, []*Channel{wildcardChannel}, false)
 24+	}()
 25+
 26+	go func() {
 27+		defer wg.Done()
 28+		_ = cast.Sub(subCtx, "sub-2", subBuf2, []*Channel{wildcardChannel}, false)
 29+	}()
 30+
 31+	time.Sleep(50 * time.Millisecond)
 32+
 33+	channel := NewChannel("logs-app1")
 34+	pubCtx, cancelPub := context.WithTimeout(context.Background(), 2*time.Second)
 35+	defer cancelPub()
 36+
 37+	_ = cast.Pub(pubCtx, "pub-1", &Buffer{b: *bytes.NewBufferString("app1-log\n")}, []*Channel{channel}, false)
 38+
 39+	time.Sleep(100 * time.Millisecond)
 40+	cancelSub()
 41+	wg.Wait()
 42+
 43+	if subBuf1.String() != "app1-log\n" {
 44+		t.Errorf("sub-1 expected app1-log, got %q", subBuf1.String())
 45+	}
 46+	if subBuf2.String() != "app1-log\n" {
 47+		t.Errorf("sub-2 expected app1-log, got %q", subBuf2.String())
 48+	}
 49+}
 50+
 51+// TestWildcardSubVariousPatterns verifies prefix, suffix, and middle asterisk wildcard matching.
 52+func TestWildcardSubVariousPatterns(t *testing.T) {
 53+	cast := NewMulticast(slog.Default())
 54+
 55+	prefixBuf := new(Buffer)
 56+	suffixBuf := new(Buffer)
 57+	middleBuf := new(Buffer)
 58+
 59+	ctx, cancel := context.WithCancel(context.Background())
 60+	defer cancel()
 61+
 62+	var wg sync.WaitGroup
 63+	wg.Add(3)
 64+
 65+	go func() {
 66+		defer wg.Done()
 67+		_ = cast.Sub(ctx, "sub-prefix", prefixBuf, []*Channel{NewChannel("metric-*")}, false)
 68+	}()
 69+	go func() {
 70+		defer wg.Done()
 71+		_ = cast.Sub(ctx, "sub-suffix", suffixBuf, []*Channel{NewChannel("*-drain")}, false)
 72+	}()
 73+	go func() {
 74+		defer wg.Done()
 75+		_ = cast.Sub(ctx, "sub-middle", middleBuf, []*Channel{NewChannel("metric-*-drain")}, false)
 76+	}()
 77+
 78+	time.Sleep(50 * time.Millisecond)
 79+
 80+	pubCtx, cancelPub := context.WithTimeout(context.Background(), 2*time.Second)
 81+	defer cancelPub()
 82+
 83+	// Publish to metric-app-drain
 84+	_ = cast.Pub(pubCtx, "pub", &Buffer{b: *bytes.NewBufferString("event\n")}, []*Channel{NewChannel("metric-app-drain")}, false)
 85+
 86+	time.Sleep(100 * time.Millisecond)
 87+	cancel()
 88+	wg.Wait()
 89+
 90+	if prefixBuf.String() != "event\n" {
 91+		t.Errorf("prefix subscriber expected event, got %q", prefixBuf.String())
 92+	}
 93+	if suffixBuf.String() != "event\n" {
 94+		t.Errorf("suffix subscriber expected event, got %q", suffixBuf.String())
 95+	}
 96+	if middleBuf.String() != "event\n" {
 97+		t.Errorf("middle subscriber expected event, got %q", middleBuf.String())
 98+	}
 99+}
100+
101+// TestWildcardSubLiteralCharsWithoutStar verifies that '?' or '[' without '*' are treated as literal names.
102+func TestWildcardSubLiteralCharsWithoutStar(t *testing.T) {
103+	cast := NewMulticast(slog.Default())
104+
105+	subBuf := new(Buffer)
106+	ctx, cancel := context.WithCancel(context.Background())
107+	defer cancel()
108+
109+	var wg sync.WaitGroup
110+	wg.Add(1)
111+
112+	// Subscribe to a topic with a literal '?' character
113+	go func() {
114+		defer wg.Done()
115+		_ = cast.Sub(ctx, "sub-literal", subBuf, []*Channel{NewChannel("topic?one")}, false)
116+	}()
117+
118+	time.Sleep(50 * time.Millisecond)
119+
120+	pubCtx, cancelPub := context.WithTimeout(context.Background(), 2*time.Second)
121+	defer cancelPub()
122+
123+	// Publish to topicXone (should NOT match because '?' is not treated as a wildcard)
124+	_ = cast.Pub(pubCtx, "pub", &Buffer{b: *bytes.NewBufferString("data\n")}, []*Channel{NewChannel("topicXone")}, false)
125+
126+	time.Sleep(100 * time.Millisecond)
127+	cancel()
128+	wg.Wait()
129+
130+	if subBuf.String() != "" {
131+		t.Errorf("literal subscriber should not have matched topicXone, got: %q", subBuf.String())
132+	}
133+}