Function testWorkerPool

pkg/workerpool/workerpool_test.go:126–210

(t *testing.T, minWorkers int, maxWorkers int, queueSize int, workerIdleTimeout time.Duration)

}

func testWorkerPool(t *testing.T, minWorkers int, maxWorkers int, queueSize int, workerIdleTimeout time.Duration) {
	a, ctx := test.New(t)
	ctx, cancel := context.WithCancel(ctx)
	defer cancel()

	workCtx := context.WithValue(ctx, "foo", "bar")

	var workToBeDone, workDone, workFailed, duplicatedWork sync.Map
	handlerCalls := int32(0)
	handler := func(ctx context.Context, item int) {
		atomic.AddInt32(&handlerCalls, 1)
		a.So(ctx, should.HaveParentContextOrEqual, workCtx)
		if item == -1 {
			panic("boom")
		}
		if _, exists := workDone.LoadOrStore(item, 0); exists {
			duplicatedWork.Store(item, 0)
		}
	}

	wp := workerpool.NewWorkerPool(workerpool.Config[int]{
		Component:         &mockComponent{},
		Context:           ctx,
		Handler:           handler,
		MinWorkers:        minWorkers,
		MaxWorkers:        maxWorkers,
		QueueSize:         queueSize,
		WorkerIdleTimeout: workerIdleTimeout,
	})

	totalWork := 100_000
	expectedHandlerCalls := int32(0)
	for i := range totalWork {
		if err := wp.Publish(workCtx, i); err != nil {
			workFailed.Store(i, 0)
		} else {
			workToBeDone.Store(i, 0)
			expectedHandlerCalls++
		}

		if rand.Intn(100) < 5 {
			if err := wp.Publish(workCtx, -1); err == nil {
				expectedHandlerCalls++
			}
		}
	}

	time.Sleep(testTimeout)
	cancel()
	wp.Wait()

	var countDone, countToBeDone, countFailed int
	workDone.Range(func(k, v any) bool {
		_, failed := workFailed.Load(k)
		a.So(failed, should.BeFalse)

		_, toBeDone := workToBeDone.Load(k)
		a.So(toBeDone, should.BeTrue)