AynaLivePlayer/pkg/eventbus/bus_impl_test.go

// generated by chatgpt
package eventbus

import (
	"fmt"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/stretchr/testify/require"
)

// TestBasicLifecycle verifies the fundamental Start, Stop, and Wait operations.
func TestBasicLifecycle(t *testing.T) {
	b := New(WithWorkerSize(2), WithQueueSize(10))

	// Start should only work once.
	err := b.Start()
	require.NoError(t, err)
	err = b.Start()
	require.NoError(t, err) // Subsequent starts should be no-ops.

	// Stop should work.
	err = b.Stop()
	require.NoError(t, err)

	// Wait should not block after stop.
	err = b.Wait()
	require.NoError(t, err)
}

// TestSubscribeAndPublish verifies the core functionality of publishing an event
// and having a subscriber receive it.
func TestSubscribeAndPublish(t *testing.T) {
	b := New(WithWorkerSize(1), WithQueueSize(10))
	err := b.Start()
	require.NoError(t, err)
	defer b.Stop()

	var wg sync.WaitGroup
	wg.Add(1)

	receivedData := new(atomic.Value)

	handler := func(event *Event) {
		receivedData.Store(event.Data)
		wg.Done()
	}

	err = b.Subscribe("", "test-event", "test-handler", handler)
	require.NoError(t, err)

	b.Publish("test-event", "hello world")

	wg.Wait()
	require.Equal(t, "hello world", receivedData.Load())
}

// TestUnsubscribe ensures that a handler stops receiving events after unsubscribing.
func TestUnsubscribe(t *testing.T) {
	b := New(WithWorkerSize(2), WithQueueSize(10))
	b.Start()
	defer b.Stop()

	var callCount int32
	handler := func(event *Event) {
		atomic.AddInt32(&callCount, 1)
	}

	err := b.Subscribe("", "event-A", "handler-1", handler)
	require.NoError(t, err)

	b.Publish("event-A", nil)
	b.Wait()
	require.Equal(t, int32(1), atomic.LoadInt32(&callCount))

	// Unsubscribe
	err = b.Unsubscribe("event-A", "handler-1")
	require.NoError(t, err)

	// Publish again
	b.Publish("event-A", nil)
	b.Wait() // Give it a moment to ensure it's not processed
	time.Sleep(50 * time.Millisecond)
	require.Equal(t, int32(1), atomic.LoadInt32(&callCount), "Handler should not be called after unsubscribing")
}

// TestSubscribeOnce verifies that a handler subscribed with SubscribeOnce
// is only called once and then automatically removed.
func TestSubscribeOnce(t *testing.T) {
	b := New(WithWorkerSize(1), WithQueueSize(10))
	b.Start()
	defer b.Stop()

	var callCount int32
	var wg sync.WaitGroup
	wg.Add(1)

	handler := func(event *Event) {
		atomic.AddInt32(&callCount, 1)
		wg.Done()
	}

	err := b.SubscribeOnce("event-once", "handler-once", handler)
	require.NoError(t, err)

	// Publish twice
	b.Publish("event-once", "data1")
	wg.Wait() // Wait for the first event to be processed

	b.Publish("event-once", "data2")
	b.Wait()
	time.Sleep(50 * time.Millisecond) // Ensure no more events are processed

	require.Equal(t, int32(1), atomic.LoadInt32(&callCount))
}

// TestChannelSubscription validates that handlers correctly receive events based on channel matching.
func TestChannelSubscription(t *testing.T) {
	b := New(WithWorkerSize(2), WithQueueSize(20))
	b.Start()
	defer b.Stop()

	var receivedMu sync.Mutex
	received := make(map[string]int)

	// Handler for a specific channel
	err := b.Subscribe("ch1", "event-X", "handler-ch1", func(event *Event) {
		receivedMu.Lock()
		received["handler-ch1"]++
		receivedMu.Unlock()
	})
	require.NoError(t, err)

	// Handler for another channel
	err = b.Subscribe("ch2", "event-X", "handler-ch2", func(event *Event) {
		receivedMu.Lock()
		received["handler-ch2"]++
		receivedMu.Unlock()
	})
	require.NoError(t, err)

	// Handler for any channel (broadcast)
	err = b.SubscribeAny("event-X", "handler-any", func(event *Event) {
		receivedMu.Lock()
		received["handler-any"]++
		receivedMu.Unlock()
	})
	require.NoError(t, err)

	// 1. Publish to ch1
	b.PublishEvent(&Event{Id: "event-X", Channel: "ch1"})
	b.Wait()
	time.Sleep(50 * time.Millisecond)
	receivedMu.Lock()
	require.Equal(t, 1, received["handler-ch1"])
	require.Equal(t, 0, received["handler-ch2"])
	require.Equal(t, 1, received["handler-any"])
	receivedMu.Unlock()

	// 2. Publish to ch2
	b.PublishEvent(&Event{Id: "event-X", Channel: "ch2"})
	b.Wait()
	time.Sleep(50 * time.Millisecond)
	receivedMu.Lock()
	require.Equal(t, 1, received["handler-ch1"])
	require.Equal(t, 1, received["handler-ch2"])
	require.Equal(t, 2, received["handler-any"])
	receivedMu.Unlock()

	// 3. Publish broadcast (empty channel)
	b.PublishEvent(&Event{Id: "event-X", Channel: ""})
	b.Wait()
	time.Sleep(50 * time.Millisecond)
	receivedMu.Lock()
	// All handlers should receive broadcast events
	require.Equal(t, 2, received["handler-ch1"])
	require.Equal(t, 2, received["handler-ch2"])
	require.Equal(t, 3, received["handler-any"])
	receivedMu.Unlock()
}

// TestPublishBeforeStart ensures that events published before the bus starts are queued
// and processed after Start() is called.
func TestPublishBeforeStart(t *testing.T) {
	b := New(WithWorkerSize(1), WithQueueSize(10))

	var receivedCount int32
	var wg sync.WaitGroup
	wg.Add(2)

	handler := func(event *Event) {
		atomic.AddInt32(&receivedCount, 1)
		wg.Done()
	}

	err := b.Subscribe("", "pending-event", "handler", handler)
	require.NoError(t, err)

	// Publish before start
	b.Publish("pending-event", "data1")
	b.Publish("pending-event", "data2")

	// Handler should not have been called yet
	require.Equal(t, int32(0), atomic.LoadInt32(&receivedCount))

	// Now start the bus
	b.Start()
	defer b.Stop()

	wg.Wait()
	require.Equal(t, int32(2), atomic.LoadInt32(&receivedCount))
}

// TestCall validates the request-response pattern using the Call method.
func TestCall(t *testing.T) {
	b := New(WithWorkerSize(2), WithQueueSize(10))
	b.Start()
	defer b.Stop()

	// Responder handler
	responder := func(event *Event) {
		require.NotEmpty(t, event.EchoId)
		// Respond with a different event ID, but the same EchoId
		b.PublishEvent(&Event{
			Id:     "response-event",
			EchoId: event.EchoId,
			Data:   fmt.Sprintf("response to %v", event.Data),
		})
	}

	err := b.Subscribe("", "request-event", "responder-handler", responder)
	require.NoError(t, err)

	// Make the call
	resp, err := b.Call("request-event", "my-data", "response-event")

	// Verify response
	require.NoError(t, err)
	require.NotNil(t, resp)
	require.Equal(t, "response-event", resp.Id)
	require.Equal(t, "response to my-data", resp.Data)
}

// TestCall_StopDuringWait checks that Call returns an error if the bus is stopped while waiting.
func TestCall_StopDuringWait(t *testing.T) {
	b := New(WithWorkerSize(1), WithQueueSize(10))
	b.Start()

	var callErr error
	var wg sync.WaitGroup
	wg.Add(1)

	go func() {
		defer wg.Done()
		// This call will never get a response
		_, callErr = b.Call("no-reply-event", nil, "no-reply-response")
	}()

	// Give the goroutine time to start waiting
	time.Sleep(100 * time.Millisecond)
	b.Stop() // Stop the bus
	wg.Wait()

	require.Error(t, callErr)
	require.Contains(t, callErr.Error(), "bus stopped")
}

// TestPanicRecovery ensures that a panicking handler does not crash the worker.
func TestPanicRecovery(t *testing.T) {
	b := New(WithWorkerSize(1), WithQueueSize(10))
	b.Start()
	defer b.Stop()

	var wg sync.WaitGroup
	wg.Add(2) // One for the panic, one for the healthy one

	panicHandler := func(event *Event) {
		defer wg.Done()
		if event.Data == "panic" {
			panic("handler intended panic")
		}
	}

	healthyHandler := func(event *Event) {
		defer wg.Done()
		// This should still run
	}

	err := b.Subscribe("", "panic-event", "panic-handler", panicHandler)
	require.NoError(t, err)
	err = b.Subscribe("", "panic-event", "healthy-handler", healthyHandler)
	require.NoError(t, err)

	// This should not crash the test
	b.Publish("panic-event", "panic")

	wg.Wait() // Will complete if both handlers finish (one by panicking, one normally)
	require.True(t, true, "Test completed, indicating panic was recovered")
}

// TestConcurrency runs many operations in parallel to check for race conditions.
func TestConcurrency(t *testing.T) {
	workerCount := 4
	queueSize := 50
	b := New(WithWorkerSize(workerCount), WithQueueSize(queueSize))
	b.Start()
	defer b.Stop()

	numGoroutines := 50
	eventsPerGoRoutine := 100
	var totalEventsPublished int32
	var totalEventsReceived int32
	var wg sync.WaitGroup

	// Subscriber goroutines
	for i := 0; i < numGoroutines; i++ {
		eventId := fmt.Sprintf("concurrent-event-%d", i%10) // 10 different events
		handlerId := fmt.Sprintf("handler-%d", i)
		wg.Add(1)
		go func() {
			defer wg.Done()
			handler := func(e *Event) {
				atomic.AddInt32(&totalEventsReceived, 1)
			}
			err := b.Subscribe("", eventId, handlerId, handler)
			require.NoError(t, err)

			time.Sleep(10 * time.Millisecond)

			err = b.Unsubscribe(eventId, handlerId)
			require.NoError(t, err)
		}()
	}

	// Setup a persistent handler to count received events
	persistentHandler := func(e *Event) {
		atomic.AddInt32(&totalEventsReceived, 1)
	}

	for i := 0; i < 20; i++ {
		eventId := fmt.Sprintf("event-for-%d", i)
		err := b.Subscribe("", eventId, "persistent-handler", persistentHandler)
		require.NoError(t, err)
	}

	// Publisher goroutines
	for i := 0; i < numGoroutines; i++ {
		wg.Add(1)
		go func(gId int) {
			defer wg.Done()
			for j := 0; j < eventsPerGoRoutine; j++ {
				eventId := fmt.Sprintf("event-for-%d", (gId+j)%20)
				b.Publish(eventId, gId)
				atomic.AddInt32(&totalEventsPublished, 1)
			}
		}(i)
	}

	wg.Wait()
	b.Wait() // Wait for all published events to be processed

	fmt.Printf("Published: %d, Received: %d\n", totalEventsPublished, totalEventsReceived)
	// We check that the number of received events matches published events for persistent handlers
	require.Equal(t, atomic.LoadInt32(&totalEventsPublished), atomic.LoadInt32(&totalEventsReceived))
}

// TestInvalidArguments checks that API methods return errors on invalid input.
func TestInvalidArguments(t *testing.T) {
	b := New(WithWorkerSize(1), WithQueueSize(1))

	// Subscribe
	err := b.Subscribe("", "", "name", func(e *Event) {})
	require.Error(t, err, "Subscribe should error on empty eventId")
	err = b.Subscribe("", "id", "", func(e *Event) {})
	require.Error(t, err, "Subscribe should error on empty handlerName")
	err = b.Subscribe("", "id", "name", nil)
	require.Error(t, err, "Subscribe should error on nil handler func")

	// SubscribeAny
	err = b.SubscribeAny("", "name", func(e *Event) {})
	require.Error(t, err, "SubscribeAny should error on empty eventId")

	// SubscribeOnce
	err = b.SubscribeOnce("", "name", func(e *Event) {})
	require.Error(t, err, "SubscribeOnce should error on empty eventId")

	// Unsubscribe
	err = b.Unsubscribe("", "name")
	require.Error(t, err, "Unsubscribe should error on empty eventId")
	err = b.Unsubscribe("id", "")
	require.Error(t, err, "Unsubscribe should error on empty handlerName")

	// Call
	_, err = b.Call("", nil, "subID")
	require.Error(t, err, "Call should error on empty eventId")
}