Function TestConcurrentConsumption

pkg/utils/incrementaldispatcher_test.go:50–97

(t *testing.T)

}

func TestConcurrentConsumption(t *testing.T) {
	producer := utils.NewIncrementalDispatcher[int]()
	numConsumers := 100
	sums := make([]atomic.Int32, numConsumers)
	var wg sync.WaitGroup

	for i := range numConsumers {
		wg.Add(1)
		i := i
		go func() {
			defer wg.Done()
			producer.ForEach(func(item int) {
				sums[i].Add(int32(item))
			})
		}()
	}

	// Add items
	expectedSum := 0
	for i := range 20 {
		expectedSum += i
		producer.Add(i)
	}

	for i := range numConsumers {
		testutils.WithTimeout(t, func() string {
			if sums[i].Load() != int32(expectedSum) {
				return fmt.Sprintf("consumer %d did not consume all the items. expected %d, actual: %d",
					i, expectedSum, sums[i].Load())
			}
			return ""
		}, time.Second)
	}

	// keep adding and ensure it's consumed
	for i := 20; i < 30; i++ {
		expectedSum += i
		producer.Add(i)
	}

	// wait for all consumers to finish
	producer.Done()
	wg.Wait()

	for i := range numConsumers {
		require.Equal(t, int32(expectedSum), sums[i].Load(), "consumer %d did not match", i)
	}
}