pizzasql/pkg/storage/notify_test.go

package storage

import (
	"testing"
	"time"
)

func TestHubSubscribePublishUnsubscribe(t *testing.T) {
	h := NewHub()
	a := NewSubscriber(1, 4)
	b := NewSubscriber(2, 4)
	h.Subscribe(a, "orders")
	h.Subscribe(a, "orders") // idempotent
	h.Subscribe(b, "other")

	h.Publish(Notification{PID: 9, Channel: "orders", Payload: "one"})
	select {
	case n := <-a.Notifications():
		if n.Payload != "one" || n.LSN == 0 {
			t.Fatalf("unexpected notification: %+v", n)
		}
	case <-time.After(time.Second):
		t.Fatal("notification not delivered")
	}
	select {
	case n := <-b.Notifications():
		t.Fatalf("received notification for another channel: %+v", n)
	default:
	}

	h.Unsubscribe(a, "orders")
	h.Publish(Notification{Channel: "orders", Payload: "two"})
	select {
	case n := <-a.Notifications():
		t.Fatalf("received notification after unsubscribe: %+v", n)
	case <-time.After(50 * time.Millisecond):
	}
}

func TestHubSlowConsumerIsDropped(t *testing.T) {
	h := NewHub()
	s := NewSubscriber(1, 2)
	h.Subscribe(s, "ch")

	for i := 0; i < 5; i++ {
		h.Publish(Notification{Channel: "ch", Payload: "x"})
	}
	select {
	case <-s.Overflow():
	case <-time.After(time.Second):
		t.Fatal("overflow signal not closed")
	}
	if channels := h.Channels(); len(channels) != 0 {
		t.Fatalf("slow consumer still subscribed: %v", channels)
	}
	published, dropped := h.Stats()
	if published != 2 || dropped == 0 {
		t.Fatalf("published=%d dropped=%d", published, dropped)
	}
}

func TestHubsAreIsolated(t *testing.T) {
	one := NewHub()
	two := NewHub()
	s := NewSubscriber(1, 2)
	one.Subscribe(s, "ch")
	two.Publish(Notification{Channel: "ch", Payload: "nope"})
	select {
	case n := <-s.Notifications():
		t.Fatalf("cross-hub notification: %+v", n)
	case <-time.After(50 * time.Millisecond):
	}
}

func TestNotificationLSNMonotonicAcrossNotifyOnlyPublishes(t *testing.T) {
	h := NewHub()
	s := NewSubscriber(1, 8)
	h.Subscribe(s, "ch")
	h.Publish(Notification{Channel: "ch", Payload: "a"})
	h.Publish(Notification{Channel: "ch", Payload: "b", LSN: 1000})
	h.Publish(Notification{Channel: "ch", Payload: "c"})

	var got []uint64
	for i := 0; i < 3; i++ {
		got = append(got, (<-s.Notifications()).LSN)
	}
	if got[0] == 0 || got[1] != 1000 || got[2] <= got[1] {
		t.Fatalf("lsns = %v", got)
	}
}