added buffer argument to OfSeq and OfSeq2

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
+	})
+}

go.mod

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

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
+}