diff --git a/map.go b/map.go index 50bc174..665af6c 100644 --- a/map.go +++ b/map.go @@ -1,13 +1,13 @@ package channel -// Map applies mapper to all I's coming from in and sends their return values to out while preserving input order. +// MapPreserveOrder applies mapper to all I's coming from source and sends their return values to out while preserving input order. // All mappings will be done as concurrently as possible using as many threads as there are CPU cores -func Map[I, O any](source <-chan I, mapper func(I) O) (out <-chan O) { - return MapWithRunner(source, getDefaultRunner(), mapper) +func MapPreserveOrder[I, O any](source <-chan I, mapper func(I) O) (out <-chan O) { + return MapPreserveOrderWithRunner(source, getDefaultRunner(), mapper) } -// MapWithRunner behaves like Map but uses runner to spawn its routines -func MapWithRunner[I, O any](source <-chan I, runner Runner, mapper func(I) O) <-chan O { +// MapPreserveOrderWithRunner behaves like MapPreserveOrder but uses runner to spawn its routines +func MapPreserveOrderWithRunner[I, O any](source <-chan I, runner Runner, mapper func(I) O) <-chan O { out := make(chan O, cap(source)) outchannels := make(chan chan O, cap(source)) @@ -39,8 +39,31 @@ func MapWithRunner[I, O any](source <-chan I, runner Runner, mapper func(I) O) < return out } -// MapSuccessive applies mapper to all I's coming from in and sends their return values to out while preserving input order. -// All mappings will be done successively +// Map applies mapper to all I's coming from source and sends their return values to out. +// All mappings will be done as concurrently as possible using as many threads as there are CPU cores +func Map[I, O any](source <-chan I, mapper func(I) O) <-chan O { + return MapWithRunner(source, getDefaultRunner(), mapper) +} + +// MapWithRunner behaves like Map but uses runner to spawn its routines +func MapWithRunner[I, O any](source <-chan I, runner Runner, mapper func(I) O) <-chan O { + out := make(chan O, cap(source)) + + go func() { + defer close(out) + for value := range source { + value := value + runner.Run(func() { + out <- mapper(value) + }) + } + }() + + return out +} + +// MapSuccessive applies mapper to all I's coming from source and sends their return values to out while preserving input order. +// All mappings will be done successively in a single thread func MapSuccessive[I, O any](source <-chan I, mapper func(I) O) <-chan O { out := make(chan O, cap(source))