Function TestQueue_ConcurrentSend

internal/base/queue/queue_test.go:161–194

(t *testing.T)

}

func TestQueue_ConcurrentSend(t *testing.T) {
	q := New[*testMessage]("test", 100)
	defer q.Close()

	var count atomic.Int32
	q.RegisterHandler(func(ctx context.Context, msg *testMessage) error {
		count.Add(1)
		return nil
	})

	var wg sync.WaitGroup
	numGoroutines := 10
	messagesPerGoroutine := 100

	for i := range numGoroutines {
		wg.Add(1)
		go func(id int) {
			defer wg.Done()
			for j := range messagesPerGoroutine {
				q.Send(context.Background(), &testMessage{ID: id*1000 + j})
			}
		}(i)
	}

	wg.Wait()

	// Wait for processing
	time.Sleep(500 * time.Millisecond)

	expected := int32(numGoroutines * messagesPerGoroutine)
	if count.Load() != expected {
		t.Errorf("expected %d messages, got %d", expected, count.Load())
	}
}

func TestQueue_ConcurrentRegisterHandler(t *testing.T) {
	q := New[*testMessage]("test", 10)