Function TestQueue_MultipleMessages

internal/base/queue/queue_test.go:59–92

(t *testing.T)

}

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

	var count atomic.Int32
	var wg sync.WaitGroup
	numMessages := 100
	wg.Add(numMessages)

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

	for i := range numMessages {
		q.Send(context.Background(), &testMessage{ID: i})
	}

	done := make(chan struct{})
	go func() {
		wg.Wait()
		close(done)
	}()

	select {
	case <-done:
		if int(count.Load()) != numMessages {
			t.Errorf("expected %d messages, got %d", numMessages, count.Load())
		}
	case <-time.After(5 * time.Second):
		t.Fatalf("timeout: only received %d of %d messages", count.Load(), numMessages)
	}
}

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