chan_io.go

@@ -21,12 +21,9 @@ func WriteIntoDelayed[T any](ch chan<- T, delay time.Duration, values ...T) {
 
 // WriteIntoWriter reads all values from ch and writes them via fmt.Fprintln to all writers
 func WriteIntoWriter[T any](ch <-chan T, writers ...io.Writer) {
-	w := io.MultiWriter(writers...)
-	EachSuccessive(ch, func(value T) {
-		if err, ok := any(value).(error); ok {
-			fmt.Fprintln(w, err.Error())
-			return
+	for value := range ch {
+		for _, w := range writers {
+			fmt.Fprintln(w, value)
 		}
-		fmt.Fprintln(w, value)
-	})
+	}
 }

each.go

@@ -1,34 +0,0 @@
-package channel
-
-import "sync"
-
-// Each consumes all values and calls f for each of them.
-// It blocks until source is closed
-func Each[T any](source <-chan T, f func(T)) {
-	EachWithRunner(source, getDefaultRunner(), f)
-}
-
-// Each consumes all values and calls f for each of them.
-// It blocks until source is closed
-func EachWithRunner[T any](source <-chan T, runner Runner, f func(T)) {
-	wg := &sync.WaitGroup{}
-
-	for value := range source {
-		value := value
-		wg.Add(1)
-		runner.Run(func() {
-			defer wg.Done()
-			f(value)
-		})
-	}
-
-	wg.Wait()
-}
-
-// EachSuccessive consumes all values and calls f for each of them.
-// It blocks until source is closed
-func EachSuccessive[T any](source <-chan T, f func(T)) {
-	for value := range source {
-		f(value)
-	}
-}

filter.go

@@ -1,39 +0,0 @@
-package channel
-
-func FilterSuccessive[T any](source <-chan T, filter func(T) bool) <-chan T {
-	out := make(chan T, cap(source))
-
-	go func() {
-		defer close(out)
-		for value := range source {
-			if filter(value) {
-				out <- value
-			}
-		}
-	}()
-
-	return out
-}
-
-func Filter[T any](source <-chan T, filter func(T) bool) <-chan T {
-	return FilterPreserveOrderWithRunner(source, getDefaultRunner(), filter)
-}
-
-func FilterPreserveOrderWithRunner[T any](source <-chan T, runner Runner, filter func(T) bool) <-chan T {
-	type FilteredValue[T any] struct {
-		Value  T
-		Filter bool
-	}
-
-	mappedValues := MapPreserveOrderWithRunner(source, runner, func(value T) FilteredValue[T] {
-		return FilteredValue[T]{Value: value, Filter: filter(value)}
-	})
-
-	filteredValues := FilterSuccessive(mappedValues, func(filteredValue FilteredValue[T]) bool {
-		return filteredValue.Filter
-	})
-
-	return MapSuccessive(filteredValues, func(filteredValue FilteredValue[T]) T {
-		return filteredValue.Value
-	})
-}

find.go

@@ -1,38 +0,0 @@
-package channel
-
-import "context"
-
-func FindFirst[T any](source <-chan T) *T {
-	for v := range source {
-		return &v
-	}
-	return nil
-}
-
-func FindFirstAndCancel[T any](source <-chan T, cancel context.CancelFunc) *T {
-	defer cancel()
-	for v := range source {
-		return &v
-	}
-	return nil
-}
-
-func FindLast[T any](source <-chan T) *T {
-	var last *T = new(T)
-	found := false
-
-	for v := range source {
-		*last = v
-		found = true
-	}
-
-	if !found {
-		return nil
-	}
-
-	return last
-}
-
-func HasAny[T any](source <-chan T) bool {
-	return FindFirst(source) != nil
-}

flat.go

@@ -1,46 +0,0 @@
-package channel
-
-func FlatSlice[T any](source <-chan []T) <-chan T {
-	out := make(chan T, cap(source))
-
-	go func() {
-		defer close(out)
-		for slice := range source {
-			for _, v := range slice {
-				out <- v
-			}
-		}
-	}()
-
-	return out
-}
-
-func FlatMap[K comparable, V, T any](source <-chan map[K]V, unmapper func(key K, value V) T) <-chan T {
-	out := make(chan T, cap(source))
-
-	go func() {
-		defer close(out)
-		for slice := range source {
-			for k, v := range slice {
-				out <- unmapper(k, v)
-			}
-		}
-	}()
-
-	return out
-}
-
-func FlatChan[T any](source <-chan <-chan T) <-chan T {
-	out := make(chan T, cap(source))
-
-	go func() {
-		defer close(out)
-		for ch := range source {
-			for v := range ch {
-				out <- v
-			}
-		}
-	}()
-
-	return out
-}

go.mod

@@ -1,3 +1,3 @@
-module git.milar.in/milarin/channel
+module git.tordarus.net/Tordarus/channel
 
-go 1.23
+go 1.18

internal_stuff.go

@@ -1,12 +0,0 @@
-package channel
-
-import "runtime"
-
-type mapEntry[K comparable, V any] struct {
-	Key   K
-	Value V
-}
-
-func getDefaultRunner() Runner {
-	return NewLimitedRunner(runtime.NumCPU())
-}

limited_runner.go

@@ -22,7 +22,7 @@ func NewLimitedRunner(routineLimit int) *LimitedRunner {
 func (r *LimitedRunner) Run(f func()) {
 	r.limiter <- struct{}{}
 	go func() {
-		defer func() { <-r.limiter }()
 		f()
+		<-r.limiter
 	}()
 }

map.go

@@ -1,15 +1,13 @@
 package channel
 
-import "sync"
-
-// 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 MapPreserveOrder[I, O any](source <-chan I, mapper func(I) O) (out <-chan O) {
-	return MapPreserveOrderWithRunner(source, getDefaultRunner(), mapper)
+// MapParallel applies mapper to all I's coming from in and sends their return values to out while conserving input order.
+// All mappings will be done as concurrently as possible
+func MapParallel[I, O any](source <-chan I, mapper func(I) O) (out <-chan O) {
+	return MapParallelWithRunner(source, NewUnlimitedRunner(), mapper)
 }
 
-// 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 {
+// MapParallelWithRunner behaves like MapParallel but uses runner to spawn its routines
+func MapParallelWithRunner[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))
 
@@ -41,38 +39,9 @@ func MapPreserveOrderWithRunner[I, O any](source <-chan I, runner Runner, mapper
 	return out
 }
 
-// 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
+// Map applies mapper to all I's coming from in and sends their return values to out while conserving input order.
+// All mappings will be done successively
 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)
-		wg := &sync.WaitGroup{}
-
-		for value := range source {
-			value := value
-			wg.Add(1)
-			runner.Run(func() {
-				defer wg.Done()
-				out <- mapper(value)
-			})
-		}
-
-		wg.Wait()
-	}()
-
-	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))
 
 	go func() {

of.go

@@ -2,7 +2,6 @@ package channel
 
 import (
 	"context"
-	"iter"
 	"time"
 )
 
@@ -52,48 +51,3 @@ func OfFunc[T any](ctx context.Context, buffer int, f func() T) <-chan T {
 
 	return out
 }
-
-// OfMap returns a channel containing the return values of the unmapper function
-// applied to any key-value pair in m
-// The order is random
-func OfMap[K comparable, V, T any](m map[K]V, unmapper func(K, V) T) <-chan T {
-	out := make(chan T, len(m))
-
-	go func() {
-		defer close(out)
-		for k, v := range m {
-			out <- unmapper(k, v)
-		}
-	}()
-
-	return out
-}
-
-// OfSeq returns a channel containing all values provided by the iterator
-func OfSeq[T any](seq iter.Seq[T], buffer int) <-chan T {
-	out := make(chan T, buffer)
-
-	go func() {
-		defer close(out)
-		for v := range seq {
-			out <- v
-		}
-	}()
-
-	return out
-}
-
-// OfSeq2 returns a channel containing the return values of the unmapper function
-// when provided with the values of the iterator
-func OfSeq2[K comparable, V, T any](seq iter.Seq2[K, V], buffer int, unmapper func(K, V) T) <-chan T {
-	out := make(chan T, buffer)
-
-	go func() {
-		defer close(out)
-		for key, value := range seq {
-			out <- unmapper(key, value)
-		}
-	}()
-
-	return out
-}

result.go

@@ -1,102 +0,0 @@
-package channel
-
-type Result[T any] struct {
-	value *T
-	err   error
-}
-
-func ResultOf[T any](value T, err error) Result[T] {
-	if err != nil {
-		return Result[T]{value: nil, err: err}
-	}
-
-	return Result[T]{value: &value, err: nil}
-}
-
-func WrapResultOutputFunc[I, O any](f func(I) (O, error)) func(I) Result[O] {
-	return func(i I) Result[O] { return ResultOf(f(i)) }
-}
-
-func WrapResultFunc[I, O any](f func(I) (O, error)) func(Result[I]) Result[O] {
-	resFunc := WrapResultOutputFunc(f)
-	nilValue := *new(O)
-	return func(r Result[I]) Result[O] {
-		v, err := r.Get()
-		if err != nil {
-			return ResultOf(nilValue, err)
-		}
-		return resFunc(v)
-	}
-}
-
-func (r Result[T]) Success() bool {
-	return r.err == nil
-}
-
-func (r Result[T]) Fail() bool {
-	return !r.Success()
-}
-
-func (r Result[T]) GetOrDefault(defaultValue T) T {
-	if r.Fail() {
-		return defaultValue
-	}
-
-	return *r.value
-}
-
-func (r Result[T]) Get() (T, error) {
-	if r.err != nil {
-		return *new(T), r.err
-	}
-	return *r.value, r.err
-}
-
-func (r Result[T]) GetUnsafe() T {
-	if r.err != nil {
-		panic(r.err)
-	}
-	return *r.value
-}
-
-func (r Result[T]) Err() error {
-	return r.err
-}
-
-func FilterSuccess[T any](source <-chan Result[T]) <-chan T {
-	succeeded := Filter(source, Result[T].Success)
-
-	return MapSuccessive(succeeded, func(r Result[T]) T {
-		v, _ := r.Get()
-		return v
-	})
-}
-
-func FilterFail[T any](source <-chan Result[T]) <-chan T {
-	failed := Filter(source, Result[T].Fail)
-
-	return MapSuccessive(failed, func(r Result[T]) T {
-		v, _ := r.Get()
-		return v
-	})
-}
-
-func FilterResults[T any](source <-chan Result[T]) (succeeded <-chan T, failed <-chan error) {
-	succ := make(chan T, cap(source))
-	fail := make(chan error, cap(source))
-
-	go func() {
-		defer close(succ)
-		defer close(fail)
-
-		for r := range source {
-			if r.Fail() {
-				fail <- r.Err()
-				continue
-			}
-			succ <- r.GetUnsafe()
-		}
-	}()
-
-	return succ, fail
-}

tee.go

@@ -1,37 +0,0 @@
-package channel
-
-// Tee returns 2 channels which both receive all values from source.
-// It's basically a copy function for channels
-func Tee[T any](source <-chan T) (<-chan T, <-chan T) {
-	outs := TeeMany(source, 2)
-	return outs[0], outs[1]
-}
-
-// TeeMany returns a given amount of channels which all receive all values from source.
-// It's basically a copy function for channels
-func TeeMany[T any](source <-chan T, amount int) []<-chan T {
-	outputs := make([]chan T, amount)
-	for i := range outputs {
-		outputs[i] = make(chan T, cap(source))
-	}
-
-	go func() {
-		defer func() {
-			for _, out := range outputs {
-				close(out)
-			}
-		}()
-
-		for value := range source {
-			for _, out := range outputs {
-				out <- value
-			}
-		}
-	}()
-
-	readOnlyOutputs := make([]<-chan T, 0, amount)
-	for _, out := range outputs {
-		readOnlyOutputs = append(readOnlyOutputs, out)
-	}
-	return readOnlyOutputs
-}

timeout.go

@@ -3,8 +3,7 @@ package channel
 import "time"
 
 // CloseOnTimeout returns a channel which receives all values from the source.
-// If no value was received in the given timeout duration, the returned channel will be closed.
-// The input channel will not be closed.
+// If no value was received in the given timeout duration, the returned channel will be closed
 func CloseOnTimeout[T any](source <-chan T, timeout time.Duration) <-chan T {
 	output := make(chan T, cap(source))
 

to.go

@@ -5,32 +5,16 @@ import "container/list"
 // ToSlice returns a slice containing all values read from ch
 func ToSlice[T any](ch <-chan T) []T {
 	s := make([]T, 0, cap(ch))
-	EachSuccessive(ch, func(value T) { s = append(s, value) })
+	Each(ch, func(value T) { s = append(s, value) })
 	return s
 }
 
-// ToSliceContinuous returns a slice containing all values read from ch.
-// The returned slice will be a pointer slice to a continuous block of memory.
-// All values will be copied.
-func ToSliceContinuous[T any](ch <-chan *T) []*T {
-	values := make([]T, 0, cap(ch))
-	pointers := make([]*T, 0, cap(ch))
-	EachSuccessive(ch, func(value *T) {
-		pointers = append(pointers, value)
-
-		if value != nil {
-			values = append(values, *value)
-		}
-	})
-	return pointers
-}
-
 // ToSliceDeref returns a slice containing all values read from ch.
 // The returned slice will be a dereferenced and continuous block of memory.
 // Nil pointers are ignored.
 func ToSliceDeref[T any](ch <-chan *T) []T {
 	s := make([]T, 0, cap(ch))
-	EachSuccessive(ch, func(value *T) {
+	Each(ch, func(value *T) {
 		if value != nil {
 			s = append(s, *value)
 		}
@@ -41,44 +25,28 @@ func ToSliceDeref[T any](ch <-chan *T) []T {
 // ToList returns a list.List containing all values read from ch
 func ToList[T any](ch <-chan T) *list.List {
 	l := list.New()
-	EachSuccessive(ch, func(value T) { l.PushBack(value) })
+	Each(ch, func(value T) { l.PushBack(value) })
 	return l
 }
 
 // ToMap returns a map containing all values read from ch.
-// The map key-value pairs are determined by f which will be called as concurrently as possible
-// to build the resulting map
-func ToMap[T any, K comparable, V any](ch <-chan T, f func(T) (K, V)) map[K]V {
-	return ToMapWithRunner(ch, getDefaultRunner(), f)
-}
-
-// ToMap returns a map containing all values read from ch.
-// The map key-value pairs are determined by f which will be called as concurrently as possible
-// to build the resulting map
-func ToMapWithRunner[T any, K comparable, V any](ch <-chan T, runner Runner, f func(T) (K, V)) map[K]V {
-	map2entry := func(t T) mapEntry[K, V] {
-		k, v := f(t)
-		return mapEntry[K, V]{Key: k, Value: v}
-	}
-
-	map2kv := func(e mapEntry[K, V]) (K, V) { return e.Key, e.Value }
-
-	return ToMapSuccessive(MapWithRunner(ch, runner, map2entry), map2kv)
-}
-
-// ToMapSuccessive returns a map containing all values read from ch.
-// The map key-value pairs are determined by f
-func ToMapSuccessive[T any, K comparable, V any](ch <-chan T, f func(T) (K, V)) map[K]V {
+// The map keys are determined by f
+func ToMap[K comparable, V any](ch <-chan V, f func(V) K) map[K]V {
 	m := map[K]V{}
-	EachSuccessive(ch, func(value T) {
-		k, v := f(value)
-		m[k] = v
-	})
+	Each(ch, func(value V) { m[f(value)] = value })
+	return m
+}
+
+// ToKeyMap returns a map containing all values read from ch as keys.
+// The map values are determined by f
+func ToKeyMap[K comparable, V any](ch <-chan K, f func(K) V) map[K]V {
+	m := map[K]V{}
+	Each(ch, func(key K) { m[key] = f(key) })
 	return m
 }
 
 // ToStructMap returns a struct{} map containing all values read from ch as keys.
-// It is a shorthand for ToMap(ch, func(value T) (T, struct{}) { return value, struct{}{} })
+// It is a shorthand for ToKeyMap(ch, func(k K) struct{} { return struct{}{} })
 func ToStructMap[T comparable](ch <-chan T) map[T]struct{} {
-	return ToMap(ch, func(value T) (T, struct{}) { return value, struct{}{} })
+	return ToKeyMap(ch, func(key T) struct{} { return struct{}{} })
 }

utils.go

@@ -5,11 +5,53 @@ func determineBufferSize[T any](channels []<-chan T) int {
 		return 0
 	}
 
-	maxBufSize := 0
+	bufSize := 0
 	for _, ch := range channels {
-		if cap(ch) > maxBufSize {
-			maxBufSize = cap(ch)
-		}
+		bufSize += cap(ch)
 	}
-	return maxBufSize
+	return bufSize / len(channels)
+}
+
+// Flush consumes all values and discards them immediately.
+// It blocks until all sources are closed
+func Flush[T any](sources ...<-chan T) {
+	for range Merge(sources...) {
+	}
+}
+
+// Each consumes all values and calls f for each of them.
+// It blocks until all sources are closed
+func Each[T any](source <-chan T, f func(T)) {
+	for value := range source {
+		f(value)
+	}
+}
+
+// Tee returns 2 channels which both receive all values from source.
+// Its basically a copy function for channels
+func Tee[T any](source <-chan T) (<-chan T, <-chan T) {
+	outs := TeeMany(source, 2)
+	return outs[0], outs[1]
+}
+
+// TeeMany returns a given amount of channels which all receive all values from source.
+// Its basically a copy function for channels
+func TeeMany[T any](source <-chan T, amount int) []<-chan T {
+	outs := make([]chan T, amount)
+
+	go func() {
+		defer func() {
+			for _, out := range outs {
+				close(out)
+			}
+		}()
+
+		for value := range source {
+			for _, out := range outs {
+				out <- value
+			}
+		}
+	}()
+
+	return (interface{}(outs)).([]<-chan T)
 }