Function newQueue

modules/distributor/queue/queue_test.go:16–35

(t *testing.T, size, workerCount int, processFunc ProcessFunc[T])

)

func newQueue[T any](t *testing.T, size, workerCount int, processFunc ProcessFunc[T]) *Queue[T] {
	cfg := Config{Name: "testName", TenantID: "testTenantID", Size: size, WorkerCount: workerCount}

	logger := log.NewNopLogger()
	q := New(cfg, logger, processFunc)

	t.Cleanup(func() {
		ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
		defer cancel()

		require.NoError(t, q.Shutdown(ctx))

		// Metrics are defined on package-level, we need to reset them each time.
		pushesTotalMetrics.Reset()
		pushesFailuresTotalMetric.Reset()
		lengthMetric.Reset()
	})

	return q
}

func newStartedQueue[T any](t *testing.T, size, workerCount int, processFunc ProcessFunc[T]) *Queue[T] {
	q := newQueue(t, size, workerCount, processFunc)