pipe: Add Pipe.DoFunnel() to handle loops over Do()

pipe/funnel.go

@@ -0,0 +1,15 @@
+package pipe
+
+// DoFunnel runs Do() for each provided slice of args.
+// Returns a slice of all output slices, or the first error encountered.
+func (p Pipe) DoFunnel(multiArgs [][]interface{}) ([][]interface{}, error) {
+	results := make([][]interface{}, len(multiArgs))
+	for i, args := range multiArgs {
+		out, err := p.Do(args...)
+		if err != nil {
+			return nil, err
+		}
+		results[i] = out
+	}
+	return results, nil
+}

pipe/funnel_test.go

@@ -0,0 +1,47 @@
+package pipe
+
+import (
+	"reflect"
+	"testing"
+)
+
+func TestFunnel(t *testing.T) {
+	p := New(Options{}).Append(func(args []interface{}) (int, error) {
+		in := args[0].(int)
+		return in, CheckErrorf(in == 5, "input was 5")
+	})
+
+	var multiArgs [][]interface{}
+	for i := 0; i < 10; i++ {
+		multiArgs = append(multiArgs, []interface{}{
+			i,
+		})
+	}
+
+	t.Run("success", func(t *testing.T) {
+		results, err := p.DoFunnel(multiArgs[:5])
+		if err != nil {
+			t.Error("Unexpected error:", err)
+		}
+		expect := [][]interface{}{
+			{0},
+			{1},
+			{2},
+			{3},
+			{4},
+		}
+		if !reflect.DeepEqual(expect, results) {
+			t.Errorf("%v != %v", expect, results)
+		}
+	})
+
+	t.Run("failed", func(t *testing.T) {
+		results, err := p.DoFunnel(multiArgs)
+		if err == nil || err.Error() != "pipe: input was 5" {
+			t.Error("Expected error to be 'input was 5', got:", err)
+		}
+		if results != nil {
+			t.Error("Expected nil results, got:", results)
+		}
+	})
+}