455 lines
12 KiB
Go
455 lines
12 KiB
Go
package bus
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
)
|
|
|
|
// recvWithTimeout blocks until a value is delivered or the deadline passes,
|
|
// failing the test on timeout. It makes delivery assertions deterministic
|
|
// instead of relying on a bare sleep.
|
|
func recvWithTimeout[T any](t *testing.T, ch <-chan T, timeout time.Duration) T {
|
|
t.Helper()
|
|
select {
|
|
case v := <-ch:
|
|
return v
|
|
case <-time.After(timeout):
|
|
t.Fatalf("timed out waiting for a delivery after %v", timeout)
|
|
var zero T
|
|
return zero
|
|
}
|
|
}
|
|
|
|
// assertNoRecv fails if a value arrives on ch within d. It is used for the
|
|
// negative cases where the registration is already gone, so nothing is ever
|
|
// enqueued and the channel stays empty.
|
|
func assertNoRecv[T any](t *testing.T, ch <-chan T, d time.Duration) {
|
|
t.Helper()
|
|
select {
|
|
case v := <-ch:
|
|
t.Fatalf("unexpected delivery: %v", v)
|
|
case <-time.After(d):
|
|
}
|
|
}
|
|
|
|
// waitSubs polls the registry until it holds exactly want subscriptions, with a
|
|
// deadline, so context-cancellation behavior can be asserted deterministically.
|
|
func waitSubs(t *testing.T, b *busImpl, want int) {
|
|
t.Helper()
|
|
deadline := time.Now().Add(2 * time.Second)
|
|
for {
|
|
b.mu.RLock()
|
|
got := len(b.subs)
|
|
b.mu.RUnlock()
|
|
if got == want {
|
|
return
|
|
}
|
|
if time.Now().After(deadline) {
|
|
t.Fatalf("subscriber count = %d, want %d", got, want)
|
|
}
|
|
time.Sleep(time.Millisecond)
|
|
}
|
|
}
|
|
|
|
func TestMatch_Rules(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
tests := []struct {
|
|
name string
|
|
pattern string
|
|
subject string
|
|
want bool
|
|
}{
|
|
{name: "exact match", pattern: "runtime.loop.start", subject: "runtime.loop.start", want: true},
|
|
{name: "exact mismatch", pattern: "runtime.loop.start", subject: "runtime.loop.stop", want: false},
|
|
{name: "exact subject has more tokens", pattern: "runtime.loop", subject: "runtime.loop.start", want: false},
|
|
{name: "exact subject has fewer tokens", pattern: "runtime.loop.start", subject: "runtime", want: false},
|
|
{name: "gt matches nested", pattern: "runtime.>", subject: "runtime.loop.start", want: true},
|
|
{name: "gt matches one child", pattern: "runtime.>", subject: "runtime.loop", want: true},
|
|
{name: "gt does not match parent", pattern: "runtime.>", subject: "runtime", want: false},
|
|
{name: "gt alone matches nested", pattern: ">", subject: "runtime.loop.start", want: true},
|
|
{name: "gt alone matches single", pattern: ">", subject: "runtime", want: true},
|
|
{name: "star matches one token", pattern: "runtime.*.start", subject: "runtime.loop.start", want: true},
|
|
{name: "star rejects extra token", pattern: "runtime.*.start", subject: "runtime.loop.x.start", want: false},
|
|
{name: "star rejects value mismatch", pattern: "runtime.*.start", subject: "runtime.loop.stop", want: false},
|
|
{name: "star alone matches single", pattern: "*", subject: "runtime", want: true},
|
|
}
|
|
for _, tt := range tests {
|
|
tt := tt
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
t.Parallel()
|
|
if got := match(tt.pattern, tt.subject); got != tt.want {
|
|
t.Errorf("match(%q, %q) = %v, want %v", tt.pattern, tt.subject, got, tt.want)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestPublish_ExactMatch(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
b := New()
|
|
defer b.Close()
|
|
got := make(chan any, 4)
|
|
|
|
if _, err := b.Subscribe(context.Background(), "runtime.loop.start", func(ctx context.Context, e Event) error {
|
|
got <- e.Payload
|
|
return nil
|
|
}); err != nil {
|
|
t.Fatalf("Subscribe: %v", err)
|
|
}
|
|
|
|
if err := b.Publish(context.Background(), "runtime.loop.start", "hit"); err != nil {
|
|
t.Fatalf("Publish: %v", err)
|
|
}
|
|
if p := recvWithTimeout(t, got, 2*time.Second); p != "hit" {
|
|
t.Fatalf("payload = %v, want hit", p)
|
|
}
|
|
|
|
if err := b.Publish(context.Background(), "runtime.loop.stop", "miss"); err != nil {
|
|
t.Fatalf("Publish: %v", err)
|
|
}
|
|
assertNoRecv(t, got, 100*time.Millisecond)
|
|
}
|
|
|
|
func TestPublish_GTailWildcard(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
b := New()
|
|
defer b.Close()
|
|
got := make(chan any, 4)
|
|
|
|
if _, err := b.Subscribe(context.Background(), "runtime.>", func(ctx context.Context, e Event) error {
|
|
got <- e.Payload
|
|
return nil
|
|
}); err != nil {
|
|
t.Fatalf("Subscribe: %v", err)
|
|
}
|
|
|
|
for _, subject := range []string{"runtime.loop.start", "runtime.loop", "runtime.loop.a.b"} {
|
|
if err := b.Publish(context.Background(), subject, subject); err != nil {
|
|
t.Fatalf("Publish(%q): %v", subject, err)
|
|
}
|
|
if p := recvWithTimeout(t, got, 2*time.Second); p != subject {
|
|
t.Fatalf("payload = %v, want %q", p, subject)
|
|
}
|
|
}
|
|
|
|
// "runtime.>" must not match the exact "runtime", and a different root
|
|
// must not match at all.
|
|
if err := b.Publish(context.Background(), "runtime", "miss-root"); err != nil {
|
|
t.Fatalf("Publish: %v", err)
|
|
}
|
|
if err := b.Publish(context.Background(), "other.x", "miss-other"); err != nil {
|
|
t.Fatalf("Publish: %v", err)
|
|
}
|
|
assertNoRecv(t, got, 100*time.Millisecond)
|
|
}
|
|
|
|
func TestPublish_StarWildcard(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
b := New()
|
|
defer b.Close()
|
|
got := make(chan any, 4)
|
|
|
|
if _, err := b.Subscribe(context.Background(), "runtime.*.start", func(ctx context.Context, e Event) error {
|
|
got <- e.Payload
|
|
return nil
|
|
}); err != nil {
|
|
t.Fatalf("Subscribe: %v", err)
|
|
}
|
|
|
|
if err := b.Publish(context.Background(), "runtime.loop.start", "hit"); err != nil {
|
|
t.Fatalf("Publish: %v", err)
|
|
}
|
|
if p := recvWithTimeout(t, got, 2*time.Second); p != "hit" {
|
|
t.Fatalf("payload = %v, want hit", p)
|
|
}
|
|
|
|
for _, subject := range []string{"runtime.loop.x.start", "runtime.loop.stop", "runtime.loop"} {
|
|
if err := b.Publish(context.Background(), subject, "miss"); err != nil {
|
|
t.Fatalf("Publish(%q): %v", subject, err)
|
|
}
|
|
}
|
|
assertNoRecv(t, got, 100*time.Millisecond)
|
|
}
|
|
|
|
func TestSubscribe_UnsubscribeStopsDelivery(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
b := New()
|
|
defer b.Close()
|
|
got := make(chan any, 2)
|
|
|
|
unsub, err := b.Subscribe(context.Background(), "a.b", func(ctx context.Context, e Event) error {
|
|
got <- e.Payload
|
|
return nil
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("Subscribe: %v", err)
|
|
}
|
|
|
|
if err := b.Publish(context.Background(), "a.b", "first"); err != nil {
|
|
t.Fatalf("Publish: %v", err)
|
|
}
|
|
if p := recvWithTimeout(t, got, 2*time.Second); p != "first" {
|
|
t.Fatalf("payload = %v, want first", p)
|
|
}
|
|
|
|
unsub()
|
|
if err := b.Publish(context.Background(), "a.b", "second"); err != nil {
|
|
t.Fatalf("Publish: %v", err)
|
|
}
|
|
assertNoRecv(t, got, 100*time.Millisecond)
|
|
}
|
|
|
|
func TestPublish_NonBlockingSlowHandler(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
b := New()
|
|
defer b.Close()
|
|
|
|
started := make(chan struct{})
|
|
release := make(chan struct{})
|
|
var once sync.Once
|
|
handler := func(ctx context.Context, e Event) error {
|
|
once.Do(func() { close(started) })
|
|
<-release
|
|
return nil
|
|
}
|
|
if _, err := b.Subscribe(context.Background(), "slow", handler); err != nil {
|
|
t.Fatalf("Subscribe: %v", err)
|
|
}
|
|
|
|
if err := b.Publish(context.Background(), "slow", "payload"); err != nil {
|
|
t.Fatalf("Publish: %v", err)
|
|
}
|
|
|
|
// The handler started and is still blocked on release. Because Publish has
|
|
// already returned, it did not wait for the handler to finish.
|
|
select {
|
|
case <-started:
|
|
case <-time.After(2 * time.Second):
|
|
t.Fatal("handler did not start")
|
|
}
|
|
select {
|
|
case <-release:
|
|
t.Fatal("handler finished before release was closed")
|
|
case <-time.After(50 * time.Millisecond):
|
|
}
|
|
close(release)
|
|
}
|
|
|
|
func TestBus_CloseStopsDelivery(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
b := New()
|
|
defer b.Close()
|
|
got := make(chan any, 1)
|
|
|
|
if _, err := b.Subscribe(context.Background(), "a.b", func(ctx context.Context, e Event) error {
|
|
got <- e.Payload
|
|
return nil
|
|
}); err != nil {
|
|
t.Fatalf("Subscribe: %v", err)
|
|
}
|
|
|
|
if err := b.Publish(context.Background(), "a.b", "before-close"); err != nil {
|
|
t.Fatalf("Publish: %v", err)
|
|
}
|
|
if p := recvWithTimeout(t, got, 2*time.Second); p != "before-close" {
|
|
t.Fatalf("payload = %v, want before-close", p)
|
|
}
|
|
|
|
if err := b.Close(); err != nil {
|
|
t.Fatalf("Close: %v", err)
|
|
}
|
|
if err := b.Publish(context.Background(), "a.b", "after-close"); !errors.Is(err, ErrClosed) {
|
|
t.Fatalf("Publish after Close error = %v, want ErrClosed", err)
|
|
}
|
|
assertNoRecv(t, got, 100*time.Millisecond)
|
|
}
|
|
|
|
func TestPublish_InvalidSubject(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
b := New()
|
|
defer b.Close()
|
|
|
|
for _, subject := range []string{"", ".a", "a.", "a..b"} {
|
|
err := b.Publish(context.Background(), subject, "x")
|
|
if !errors.Is(err, ErrInvalidArgument) {
|
|
t.Errorf("Publish(%q) error = %v, want ErrInvalidArgument", subject, err)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestSubscribe_InvalidPattern(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
b := New()
|
|
defer b.Close()
|
|
|
|
for _, pattern := range []string{"", ".a", "a.", "a..b", "a.>.b"} {
|
|
_, err := b.Subscribe(context.Background(), pattern, func(ctx context.Context, e Event) error { return nil })
|
|
if !errors.Is(err, ErrInvalidArgument) {
|
|
t.Errorf("Subscribe(%q) error = %v, want ErrInvalidArgument", pattern, err)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestSubscribe_NilHandler(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
b := New()
|
|
defer b.Close()
|
|
|
|
_, err := b.Subscribe(context.Background(), "a.b", nil)
|
|
if !errors.Is(err, ErrInvalidArgument) {
|
|
t.Fatalf("Subscribe(nil) error = %v, want ErrInvalidArgument", err)
|
|
}
|
|
}
|
|
|
|
func TestBus_ClosedError(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
b := New()
|
|
if err := b.Close(); err != nil {
|
|
t.Fatalf("Close: %v", err)
|
|
}
|
|
|
|
if err := b.Publish(context.Background(), "a.b", "x"); !errors.Is(err, ErrClosed) {
|
|
t.Fatalf("Publish after Close error = %v, want ErrClosed", err)
|
|
}
|
|
if _, err := b.Subscribe(context.Background(), "a.b", func(ctx context.Context, e Event) error { return nil }); !errors.Is(err, ErrClosed) {
|
|
t.Fatalf("Subscribe after Close error = %v, want ErrClosed", err)
|
|
}
|
|
if err := b.Close(); err != nil {
|
|
t.Fatalf("second Close error = %v, want nil", err)
|
|
}
|
|
}
|
|
|
|
func TestBus_DroppedCounter(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
b := New(WithQueueSize(1))
|
|
defer b.Close()
|
|
|
|
started := make(chan struct{})
|
|
release := make(chan struct{})
|
|
var once sync.Once
|
|
if _, err := b.Subscribe(context.Background(), "d", func(ctx context.Context, e Event) error {
|
|
once.Do(func() { close(started) })
|
|
<-release
|
|
return nil
|
|
}); err != nil {
|
|
t.Fatalf("Subscribe: %v", err)
|
|
}
|
|
|
|
if err := b.Publish(context.Background(), "d", "first"); err != nil {
|
|
t.Fatalf("Publish first: %v", err)
|
|
}
|
|
select {
|
|
case <-started:
|
|
case <-time.After(2 * time.Second):
|
|
t.Fatal("handler did not start; worker failed to drain")
|
|
}
|
|
|
|
// The worker is now blocked in the handler, so it consumed "first" and the
|
|
// single-slot buffer is empty. The next enqueue fills it; the two after
|
|
// that find it full and are dropped.
|
|
for _, payload := range []string{"fill1", "fill2", "fill3"} {
|
|
if err := b.Publish(context.Background(), "d", payload); err != nil {
|
|
t.Fatalf("Publish %s: %v", payload, err)
|
|
}
|
|
}
|
|
if got := b.Dropped(); got != 2 {
|
|
t.Fatalf("Dropped() = %d, want 2", got)
|
|
}
|
|
close(release)
|
|
}
|
|
|
|
func TestSubscribe_ContextCancelAutoUnsub(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
b := New()
|
|
defer b.Close()
|
|
got := make(chan any, 1)
|
|
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
if _, err := b.Subscribe(ctx, "a.b", func(ctx context.Context, e Event) error {
|
|
got <- e.Payload
|
|
return nil
|
|
}); err != nil {
|
|
t.Fatalf("Subscribe: %v", err)
|
|
}
|
|
|
|
cancel()
|
|
waitSubs(t, b, 0)
|
|
|
|
if err := b.Publish(context.Background(), "a.b", "after-cancel"); err != nil {
|
|
t.Fatalf("Publish: %v", err)
|
|
}
|
|
assertNoRecv(t, got, 100*time.Millisecond)
|
|
}
|
|
|
|
func TestPublish_ConcurrentSafety(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
b := New(WithQueueSize(8))
|
|
defer b.Close()
|
|
|
|
const workers = 8
|
|
const iterations = 200
|
|
var wg sync.WaitGroup
|
|
for w := 0; w < workers; w++ {
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
for i := 0; i < iterations; i++ {
|
|
if i%3 == 0 {
|
|
unsub, err := b.Subscribe(context.Background(), "c", func(ctx context.Context, e Event) error {
|
|
return nil
|
|
})
|
|
if err == nil {
|
|
unsub()
|
|
}
|
|
continue
|
|
}
|
|
_ = b.Publish(context.Background(), "c", i)
|
|
_ = b.Publish(context.Background(), "c.x", i)
|
|
}
|
|
}()
|
|
}
|
|
wg.Wait()
|
|
|
|
if err := b.Close(); err != nil {
|
|
t.Fatalf("Close: %v", err)
|
|
}
|
|
}
|
|
|
|
func TestClose_Concurrent(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
b := New()
|
|
var wg sync.WaitGroup
|
|
for i := 0; i < 8; i++ {
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
_ = b.Publish(context.Background(), "a.b", 1)
|
|
_, _ = b.Subscribe(context.Background(), "a.b", func(ctx context.Context, e Event) error { return nil })
|
|
_ = b.Close()
|
|
}()
|
|
}
|
|
wg.Wait()
|
|
|
|
if err := b.Close(); err != nil {
|
|
t.Fatalf("Close: %v", err)
|
|
}
|
|
}
|