added generic types + FilterChanges

each.go

@@ -4,13 +4,13 @@ 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)) {
+func Each[T any](source <-chan T, f Consumer[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)) {
+func EachWithRunner[T any](source <-chan T, runner Runner, f Consumer[T]) {
 	wg := &sync.WaitGroup{}
 
 	for value := range source {
@@ -27,7 +27,7 @@ func EachWithRunner[T any](source <-chan T, runner Runner, f func(T)) {
 
 // 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)) {
+func EachSuccessive[T any](source <-chan T, f Consumer[T]) {
 	for value := range source {
 		f(value)
 	}

filter.go

@@ -1,6 +1,16 @@
 package channel
 
-func FilterSuccessive[T any](source <-chan T, filter func(T) bool) <-chan T {
+func FilterChanges[T any](source <-chan T, cmp EqualityComparator[T]) <-chan T {
+	oldValue := new(T)
+	oldValue = nil
+	return FilterSuccessive(source, func(newValue T) bool {
+		res := oldValue == nil || !cmp(*oldValue, newValue)
+		oldValue = &newValue
+		return res
+	})
+}
+
+func FilterSuccessive[T any](source <-chan T, filter FilterFunc[T]) <-chan T {
 	out := make(chan T, cap(source))
 
 	go func() {
@@ -15,11 +25,11 @@ func FilterSuccessive[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 {
+func Filter[T any](source <-chan T, filter FilterFunc[T]) <-chan T {
 	return FilterPreserveOrderWithRunner(source, getDefaultRunner(), filter)
 }
 
-func FilterPreserveOrderWithRunner[T any](source <-chan T, runner Runner, filter func(T) bool) <-chan T {
+func FilterPreserveOrderWithRunner[T any](source <-chan T, runner Runner, filter FilterFunc[T]) <-chan T {
 	type FilteredValue[T any] struct {
 		Value  T
 		Filter bool

flat.go

@@ -15,7 +15,7 @@ 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 {
+func FlatMap[K comparable, V, T any](source <-chan map[K]V, unmapper MapUnmapper[K, V, T]) <-chan T {
 	out := make(chan T, cap(source))
 
 	go func() {

map.go

@@ -4,12 +4,12 @@ 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) {
+func MapPreserveOrder[I, O any](source <-chan I, mapper Mapper[I, O]) (out <-chan O) {
 	return MapPreserveOrderWithRunner(source, getDefaultRunner(), 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 {
+func MapPreserveOrderWithRunner[I, O any](source <-chan I, runner Runner, mapper Mapper[I, O]) <-chan O {
 	out := make(chan O, cap(source))
 	outchannels := make(chan chan O, cap(source))
 
@@ -43,12 +43,12 @@ func MapPreserveOrderWithRunner[I, O any](source <-chan I, runner Runner, mapper
 
 // 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 {
+func Map[I, O any](source <-chan I, mapper Mapper[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 {
+func MapWithRunner[I, O any](source <-chan I, runner Runner, mapper Mapper[I, O]) <-chan O {
 	out := make(chan O, cap(source))
 
 	go func() {
@@ -72,7 +72,7 @@ func MapWithRunner[I, O any](source <-chan I, runner Runner, mapper func(I) O) <
 
 // 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 {
+func MapSuccessive[I, O any](source <-chan I, mapper Mapper[I, O]) <-chan O {
 	out := make(chan O, cap(source))
 
 	go func() {

of.go

@@ -56,7 +56,7 @@ func OfFunc[T any](ctx context.Context, buffer int, f func() T) <-chan T {
 // 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 {
+func OfMap[K comparable, V, T any](m map[K]V, unmapper MapUnmapper[K, V, T]) <-chan T {
 	out := make(chan T, len(m))
 
 	go func() {
@@ -85,7 +85,7 @@ func OfSeq[T any](seq iter.Seq[T], buffer int) <-chan T {
 
 // 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 {
+func OfSeq2[K comparable, V, T any](seq iter.Seq2[K, V], buffer int, unmapper MapUnmapper[K, V, T]) <-chan T {
 	out := make(chan T, buffer)
 
 	go func() {

to.go

@@ -48,16 +48,16 @@ func ToList[T any](ch <-chan T) *list.List {
 // 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)
+func ToMap[T any, K comparable, V any](ch <-chan T, mapper MapMapper[T, K, V]) map[K]V {
+	return ToMapWithRunner(ch, getDefaultRunner(), mapper)
 }
 
 // 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 {
+func ToMapWithRunner[T any, K comparable, V any](ch <-chan T, runner Runner, mapper MapMapper[T, K, V]) map[K]V {
 	map2entry := func(t T) mapEntry[K, V] {
-		k, v := f(t)
+		k, v := mapper(t)
 		return mapEntry[K, V]{Key: k, Value: v}
 	}
 
@@ -68,10 +68,10 @@ func ToMapWithRunner[T any, K comparable, V any](ch <-chan T, runner Runner, f f
 
 // 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 {
+func ToMapSuccessive[T any, K comparable, V any](ch <-chan T, mapper MapMapper[T, K, V]) map[K]V {
 	m := map[K]V{}
 	EachSuccessive(ch, func(value T) {
-		k, v := f(value)
+		k, v := mapper(value)
 		m[k] = v
 	})
 	return m

types.go

@@ -0,0 +1,18 @@
+package channel
+
+func DefaultEqualityComparator[T comparable](a, b T) bool {
+	return a == b
+}
+
+type EqualityComparator[T any] = func(a, b T) bool
+type FilterFunc[T any] = func(T) bool
+type Consumer[T any] = func(T)
+type IndexedConsumer[T any] = func(int, T)
+
+type Mapper[I, O any] = func(I) O
+type ErrorMapper[I, O any] = func(I) (O, error)
+
+type MapMapper[T any, K comparable, V any] = func(T) (K, V)
+type MapUnmapper[K comparable, V, T any] = func(K, V) T
+
+type Reducer[A, T any] = func(acc A, value T) A