pipe function to consume stream to void (#71)

Author: fogfish

pipe/fork/fork.go

@@ -90,6 +90,36 @@ func ForEach[A any](ctx context.Context, par int, in <-chan A, f F[A, A]) <-chan
 	return done
 }
 
+// Void applies nothing for each message in the channel, making channel empty
+func Void[A any](ctx context.Context, par int, in <-chan A) <-chan struct{} {
+	var wg sync.WaitGroup
+	done := make(chan struct{})
+
+	fmap := func() {
+		defer wg.Done()
+
+		for range in {
+			select {
+			case <-ctx.Done():
+				return
+			default:
+			}
+		}
+	}
+
+	wg.Add(par)
+	for i := 1; i <= par; i++ {
+		go fmap()
+	}
+
+	go func() {
+		wg.Wait()
+		close(done)
+	}()
+
+	return done
+}
+
 // FMap applies function over channel messages, flatten the output channel and
 // emits it result to new channel.
 func FMap[A, B any](ctx context.Context, par int, in <-chan A, fmap FF[A, B]) (<-chan B, <-chan error) {

pipe/fork/fork_test.go

@@ -101,6 +101,14 @@ func TestForEach(t *testing.T) {
 	close()
 }
 
+func TestVoid(t *testing.T) {
+	ctx, close := context.WithCancel(context.Background())
+	seq := fork.Seq(1, 2, 3, 4, 5)
+	<-fork.Void(ctx, par, seq)
+
+	close()
+}
+
 func TestFMap(t *testing.T) {
 	t.Run("FMap", func(t *testing.T) {
 		fun := fork.LiftF(

pipe/pipe.go

@@ -99,6 +99,25 @@ func ForEach[A any](ctx context.Context, in <-chan A, f F[A, A]) <-chan struct{}
 	return done
 }
 
+// Void applies nothing for each message in the channel, making channel empty
+func Void[A any](ctx context.Context, in <-chan A) <-chan struct{} {
+	done := make(chan struct{})
+
+	go func() {
+		defer close(done)
+
+		for range in {
+			select {
+			case <-ctx.Done():
+				return
+			default:
+			}
+		}
+	}()
+
+	return done
+}
+
 // FMap applies function over channel messages, flatten the output channel and
 // emits it result to new channel.
 func FMap[A, B any](ctx context.Context, in <-chan A, fmap FF[A, B]) (<-chan B, <-chan error) {

pipe/pipe_test.go

@@ -108,6 +108,14 @@ func TestForEach(t *testing.T) {
 	close()
 }
 
+func TestVoid(t *testing.T) {
+	ctx, close := context.WithCancel(context.Background())
+	seq := pipe.Seq(1, 2, 3, 4, 5)
+	<-pipe.Void(ctx, seq)
+
+	close()
+}
+
 func TestFMap(t *testing.T) {
 	t.Run("FMap", func(t *testing.T) {
 		fun := pipe.LiftF(

pipe/version.go

@@ -8,4 +8,4 @@
 
 package pipe
 
-const Version = "pipe/v2.0.1"
+const Version = "pipe/v2.1.0"