Function TestWatchManager_SubscribeAndSendMessage

watch_manager_test.go:205–398

(t *testing.T)

}

func TestWatchManager_SubscribeAndSendMessage(t *testing.T) {
	t.Run("subscribe and send message", func(t *testing.T) {
		wm := startDistributor(t)
		defer func() { _ = wm.close() }()

		bucket := "test"
		key := "test"
		value := []byte("updated value")

		receiveChan, _ := wmSubscribe(t, wm, bucket, key)
		wmSendMessage(t, wm, bucket, key, value)

		msg := wmReceiveMessage(t, receiveChan, bucket, key, 1*time.Second)
		assert.Equal(t, value, msg.Value)
	})

	t.Run("subscribe and drop messages", func(t *testing.T) {
		wm := startDistributor(t)
		defer func() { _ = wm.close() }()

		bucket := "bucket_test"
		key := "key_test"

		// Subscribe but don't receive messages
		receiveChan, _ := wmSubscribe(t, wm, bucket, key)

		// Send enough messages to:
		// 1. Fill receiveChan buffer (1024)
		// 2. Trigger 100+ drops to force-unsubscribe
		// Total: 1100+ messages
		messagesToSend := 1124
		wmSendMessages(t, wm, bucket, key, messagesToSend)

		time.Sleep(100 * time.Millisecond)
		drained := wmDrainChannel(t, receiveChan, 2*time.Second)

		assert.Equal(t, 1024, drained, "should have buffered exactly 1024 messages")

		// Verify subscriber was force-unsubscribed due to slow consumption
		wmVerifySubscriberRemoved(t, wm, bucket, key)
		wmVerifyBucketRemoved(t, wm, bucket)
	})

	t.Run("multiple subscribers for the same key", func(t *testing.T) {
		wm := startDistributor(t)
		defer func() { _ = wm.close() }()

		bucket := "bucket_test"
		key := "key_test"
		numSubscribers := 10
		messagesPerSubscriber := 10

		wg := sync.WaitGroup{}

		// start subscribers goroutines for the same key
		for i := 0; i < numSubscribers; i++ {
			wg.Add(1)
			go func(i int) {