Function Run

api/concurrent/concurrent.go:94–148

Run runs a list of functions returns the results

(ctx context.Context, maxConcurrency int, tasks ...Func)

// Run runs a list of functions returns the results
func Run(ctx context.Context, maxConcurrency int, tasks ...Func) ([]Result, error) {
	var wg sync.WaitGroup

	resultsChan := make(chan Result, len(tasks))
	taskChan := make(chan Func, len(tasks))

	localCtx, cancelCtx := context.WithCancel(ctx)
	defer cancelCtx()

	runTask := func() {
		defer wg.Done()

		for fn := range taskChan {
			result, err := fn(localCtx)
			resultsChan <- Result{Result: result, Err: err}
		}
	}

	// Set maxConcurrency to the number of tasks if zero or negative
	if maxConcurrency <= 0 {
		maxConcurrency = len(tasks)
	}

	// Start worker goroutines
	for range maxConcurrency {
		wg.Add(1)
		go runTask()
	}

	// Add tasks to the task channel
	for _, fn := range tasks {
		taskChan <- fn
	}

	// Close the task channel to signal workers to stop when all tasks are done
	close(taskChan)

	// Wait for all workers to complete
	wg.Wait()
	close(resultsChan)

	// Collect the results and cancel on error
	results := make([]Result, 0, len(tasks))
	for r := range resultsChan {
		if r.Err != nil {
			cancelCtx()

			return nil, r.Err
		}

		results = append(results, r)
	}

	return results, nil
}