1package pubsub
  2
  3import (
  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.
 15func 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}
 76
 77// TestWildcardSubMultipleSubscribers verifies that multiple wildcard subscribers
 78// listening on the same pattern both receive published messages.
 79func 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.
123func 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.
173func 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}