Batch Processor in Go

batch_processor.go

// BatchProcessor processes a collection of items in batches using goroutines.
func BatchProcessor(items []int, processFunc func([]int) int, batchSize int) <-chan int {
	var wg sync.WaitGroup

	// Create a channel with a buffer size of the batch size.
	ch := make(chan []int, batchSize)

	// Create a channel to receive the results returned by the Process function.
	results := make(chan int)

	// Start a goroutine to read items from the channel and process them.
	wg.Add(1)
	go func() {
		defer wg.Done()
		for batch := range ch {
			// Call the Process function and send the result on the results channel.
			results <- processFunc(batch)
		}
		// Close the results channel when all of the batches have been processed.
		close(results)
	}()

	// Divide the collection of items into batches and send each batch on the channel.
	for i := 0; i < len(items); i += batchSize {
		end := i + batchSize
		if end > len(items) {
			end = len(items)
		}
		batch := items[i:end]
		ch <- batch
	}

	// Close the channel to signal to the goroutine that no more batches will be sent.
	close(ch)

	// Wait for the goroutine to finish processing all of the batches.
	wg.Wait()

	// Return the results channel.
	return results
}