Change Tap* function signature and always forward the T type

chan.go

@@ -73,15 +73,15 @@ func (c Chan[T]) MapWithErrorSink(size int, mp func(T) (any, error), sink func(e
 	return MapWithErrorSink(size, mp, sink, thunkChanPull(c))
 }
 
-func (c Chan[T]) Tap(size int, tap func(T) T) ChanPull[T] {
+func (c Chan[T]) Tap(size int, tap func(T)) ChanPull[T] {
 	return Tap(size, tap, thunkChanPull(c))
 }
 
-func (c Chan[T]) TapWithError(size int, tap func(T) (T, error)) (ChanPull[T], ChanPull[error]) {
+func (c Chan[T]) TapWithError(size int, tap func(T) error) (ChanPull[T], ChanPull[error]) {
 	return TapWithError(size, tap, thunkChanPull(c))
 }
 
-func (c Chan[T]) TapWithErrorSink(size int, tap func(T) (T, error), sink func(error)) ChanPull[T] {
+func (c Chan[T]) TapWithErrorSink(size int, tap func(T) error, sink func(error)) ChanPull[T] {
 	return TapWithErrorSink(size, tap, sink, thunkChanPull(c))
 }
 

chan_pull.go

@@ -52,15 +52,15 @@ func (c ChanPull[T]) MapWithErrorSink(size int, mp func(T) (any, error), sink fu
 	return MapWithErrorSink(size, mp, sink, c)
 }
 
-func (c ChanPull[T]) Tap(size int, tap func(T) T) ChanPull[T] {
+func (c ChanPull[T]) Tap(size int, tap func(T)) ChanPull[T] {
 	return Tap(size, tap, c)
 }
 
-func (c ChanPull[T]) TapWithError(size int, tap func(T) (T, error)) (ChanPull[T], ChanPull[error]) {
+func (c ChanPull[T]) TapWithError(size int, tap func(T) error) (ChanPull[T], ChanPull[error]) {
 	return TapWithError(size, tap, c)
 }
 
-func (c ChanPull[T]) TapWithErrorSink(size int, tap func(T) (T, error), sink func(error)) ChanPull[T] {
+func (c ChanPull[T]) TapWithErrorSink(size int, tap func(T) error, sink func(error)) ChanPull[T] {
 	return TapWithErrorSink(size, tap, sink, c)
 }
 

tap.go

@@ -1,13 +1,47 @@
 package pipes
 
-func Tap[T any](size int, tap func(T) T, in <-chan T) ChanPull[T] {
-	return Map(size, tap, in)
+func Tap[T any](size int, tap func(T), in <-chan T) ChanPull[T] {
+	out := make(chan T, size)
+	go tapWorker(tap, in, out)
+	return out
 }
 
-func TapWithError[T any](size int, tap func(T) (T, error), in <-chan T) (ChanPull[T], ChanPull[error]) {
-	return MapWithError(size, tap, in)
+func tapWorker[T any](tap func(T), in <-chan T, out chan<- T) {
+	defer close(out)
+	for t := range in {
+		tap(t)
+		out <- t
+	}
 }
 
-func TapWithErrorSink[T any](size int, tap func(T) (T, error), sink func(error), in <-chan T) ChanPull[T] {
-	return MapWithErrorSink(size, tap, sink, in)
+func TapWithError[T any](size int, tap func(T) error, in <-chan T) (ChanPull[T], ChanPull[error]) {
+	out, err := make(chan T, size), make(chan error, size)
+	go tapWithErrorWorker(tap, in, out, err)
+	return out, err
+}
+
+func tapWithErrorWorker[T any](tap func(T) error, in <-chan T, out chan<- T, err chan<- error) {
+	defer func() { close(out); close(err) }()
+	for t := range in {
+		if er := tap(t); er != nil {
+			err <- er
+		}
+		out <- t
+	}
+}
+
+func TapWithErrorSink[T any](size int, tap func(T) error, sink func(error), in <-chan T) ChanPull[T] {
+	out := make(chan T, size)
+	go tapWithErrorSinkWorker(tap, sink, in, out)
+	return out
+}
+
+func tapWithErrorSinkWorker[T any](mp func(T) error, sink func(error), in <-chan T, out chan<- T) {
+	defer close(out)
+	for t := range in {
+		if er := mp(t); er != nil {
+			sink(er)
+		}
+		out <- t
+	}
 }