The grouping loop
A map, a slice of keys so the output has an order, an if _, ok, and a
sort at the end. You have written it a hundred times.
One pass, memory bounded by the number of countries rather than the number of
orders, keys in first-seen order, and the chain keeps going.
// Revenue per country, largest first.
ranked := catena.FromSlice(orders).
FoldedBy(Order.Country,
func(string) int { return 0 },
func(sum int, o Order) int { return sum + o.Total }).
MapTo(func(country string, total int) Row { return Row{country, total} }).
SortedByDesc(func(r Row) int { return r.Total })
The worker pool
A semaphore, a WaitGroup, a results slice indexed so things come back in
order, an error channel, cancelling the rest when one fails — and hoping nothing
is still running when you return.
loadCustomer is an ordinary function. Every goroutine has exited by the
time Collect returns — on success, on error, on cancel, and on panic.
// func loadCustomer(ctx context.Context, id ID) (Customer, error)
customers, err := catena.FromSlice(ids).
Then(parallel.Map(ctx, 8, loadCustomer)). // 8 in flight, results in order
Collect() // first error cancels the rest
The error plumbing
Half the loop is if err != nil, and whether a bad row aborts the import or
gets skipped is decided deep inside it, where the caller cannot change it.
The pipeline carries the errors; the consumer picks the policy. The file is opened
when the pipeline runs and closed on every way out of it.
rows := source.Lines(open).MapErr(parseRow) // a failure is an element
good, err := rows.Collect() // stop at the first bad row
good, errs := rows.CollectAll() // import what parses, report the rest
good := rows.Ignore().Collect() // skip bad rows
