diff --git a/subpipe.go b/subpipe.go
index 6f69b0d..c470b37 100644
--- a/subpipe.go
+++ b/subpipe.go
@@ -17,11 +17,11 @@ func eachLine(r io.Reader, fn func(line string) error) error {
 	return scanner.Err()
 }
 
-func SubPipe(fn func(src *script.Pipe) *script.Pipe) func(r io.Reader, w io.Writer) error {
+func SubPipe(fn func(line string) *script.Pipe) func(r io.Reader, w io.Writer) error {
 	return func(r io.Reader, w io.Writer) error {
 		scanner := bufio.NewScanner(r)
 		for scanner.Scan() {
-			bts, err := fn(script.Echo(scanner.Text())).Bytes()
+			bts, err := fn(scanner.Text()).Bytes()
 			if err != nil {
 				return err
 			}
diff --git a/subpipe_test.go b/subpipe_test.go
index d96fb8a..0bcecb4 100644
--- a/subpipe_test.go
+++ b/subpipe_test.go
@@ -10,8 +10,8 @@ import (
 func TestSubPipe(t *testing.T) {
 	t.Run("should run the subpipe for each line in the parent pipe", func(t *testing.T) {
 		verifyPipeLines(t, func(p *script.Pipe) *script.Pipe {
-			return p.Filter(scriptx.SubPipe(func (src *script.Pipe) *script.Pipe {
-				return src.Replace("Line", "Subpipe")
+			return p.Filter(scriptx.SubPipe(func (line string) *script.Pipe {
+				return script.Echo(line).Replace("Line", "Subpipe")
 			}))
 		}, []string{
 			"Line 1",