added buffer argument to OfSeq and OfSeq2

chan_io.go

@@ -23,6 +23,10 @@ func WriteIntoDelayed[T any](ch chan<- T, delay time.Duration, values ...T) {
 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
+		}
 		fmt.Fprintln(w, value)
 	})
 }

filter.go

@@ -1,6 +1,6 @@
 package channel
 
-func Filter[T any](source <-chan T, filter func(T) bool) <-chan T {
+func FilterSuccessive[T any](source <-chan T, filter func(T) bool) <-chan T {
 	out := make(chan T, cap(source))
 
 	go func() {
@@ -14,3 +14,26 @@ func Filter[T any](source <-chan T, filter func(T) bool) <-chan T {
 
 	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

@@ -10,8 +10,8 @@ func FindFirst[T any](source <-chan T) *T {
 }
 
 func FindFirstAndCancel[T any](source <-chan T, cancel context.CancelFunc) *T {
+	defer cancel()
 	for v := range source {
-		cancel()
 		return &v
 	}
 	return nil

flat.go

@@ -15,6 +15,21 @@ func FlatSlice[T any](source <-chan []T) <-chan T {
 	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.mod

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

map.go

@@ -1,13 +1,15 @@
 package channel
 
-// Map applies mapper to all I's coming from in and sends their return values to out while preserving input order.
+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 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 +41,37 @@ 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)
+		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))
 

of.go

@@ -2,6 +2,7 @@ package channel
 
 import (
 	"context"
+	"iter"
 	"time"
 )
 
@@ -58,7 +59,7 @@ func OfFunc[T any](ctx context.Context, buffer int, f func() T) <-chan T {
 func OfMap[K comparable, V, T any](m map[K]V, unmapper func(K, V) T) <-chan T {
 	out := make(chan T, len(m))
 
-	defer func() {
+	go func() {
 		defer close(out)
 		for k, v := range m {
 			out <- unmapper(k, v)
@@ -67,3 +68,32 @@ func OfMap[K comparable, V, T any](m map[K]V, unmapper func(K, V) T) <-chan T {
 
 	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

@@ -0,0 +1,102 @@
+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
+}

to.go

@@ -9,6 +9,22 @@ func ToSlice[T any](ch <-chan T) []T {
 	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.