Repository: bitfield/script
Pull request #101 (2022-03-21 18:44:10)

Make pipe filters concurrent


Patch:
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
index f281ada..4dd555d 100644
--- a/CONTRIBUTING.md
+++ b/CONTRIBUTING.md
@@ -1,32 +1,5 @@
 So you'd like to contribute to the `script` library? Excellent! Thank you very much. I can absolutely use your help.
 
-- [Getting started](#getting-started)
-	- [Look for existing issues](#look-for-existing-issues)
-	- [Open a new issue before making a PR](#open-a-new-issue-before-making-a-pr)
-	- [Write a use case](#write-a-use-case)
-- [Coding standards](#coding-standards)
-	- [Tests](#tests)
-		- [Use the standard library](#use-the-standard-library)
-		- [Spend time on your test cases](#spend-time-on-your-test-cases)
-		- [Add your method to `doMethodsOnPipe` for stress testing](#add-your-method-to-domethodsonpipe-for-stress-testing)
-	- [Dealing with errors](#dealing-with-errors)
-		- [Don't panic](#dont-panic)
-		- [Set the pipe's error status](#set-the-pipes-error-status)
-	- [Style and formatting](#style-and-formatting)
-- [Documentation](#documentation)
-	- [Write doc comments](#write-doc-comments)
-	- [Update the README](#update-the-readme)
-- [Writing pipe operations](#writing-pipe-operations)
-	- [Writing a source](#writing-a-source)
-	- [Writing a filter](#writing-a-filter)
-	- [Writing a sink](#writing-a-sink)
-- [Before submitting your pull request](#before-submitting-your-pull-request)
-- [After submitting your PR](#after-submitting-your-pr)
-- [The code review process](#the-code-review-process)
-	- [Expect to be taken seriously](#expect-to-be-taken-seriously)
-	- [Dealing with comments](#dealing-with-comments)
-	- [This may take a while](#this-may-take-a-while)
-
 # Getting started
 
 Here are some hints on a good workflow for contributing to the project.
@@ -39,7 +12,7 @@ If you already have an idea for a feature you want to add, check the issues list
 
 ## Open a new issue before making a PR
 
-I _don't_ recommend just making a pull request for some new feature. Usually it's better to [open an issue](https://github.com/bitfield/script/issues/new) first, and we can discuss what the feature is about, how best to design it, other people can weigh in with contributions, and so forth. Design is, in fact, the hard part. Once we have a solid, well-thought-out design, implementing it is usually fairly easy. (Implementing a bad design may be easy too, but it's a waste of effort.)
+I _don't_ recommend just making a pull request for some new feature—it probably won't be accepted! Usually it's better to [open an issue](https://github.com/bitfield/script/issues/new) first, and we can discuss what the feature is about, how best to design it, other people can weigh in with contributions, and so forth. Design is, in fact, the hard part. Once we have a solid, well-thought-out design, implementing it is usually fairly easy. (Implementing a bad design may be easy too, but it's a waste of effort.)
 
 ## Write a use case
 
@@ -133,11 +106,11 @@ Any functions or methods you write should have useful documentation comments in
 For example:
 
 ```go
-// Freq reads from the pipe, and returns a new pipe containing only unique lines
-// from the input, prefixed with a frequency count, in descending numerical
-// order (most frequent lines first). Lines with equal frequency will be sorted
-// alphabetically. If there is an error reading the pipe, the pipe's error
-// status is also set.
+// WriteFile writes the contents of the pipe to the specified file, and closes
+// the pipe after reading. If the file already exists, it is truncated and the
+// new data will replace the old. It returns the number of bytes successfully
+// written, or an error.
+func (p *Pipe) WriteFile(fileName string) (int64, error) {
 ```
 
 This is the _whole_ user manual for your code. It will be included in the autogenerated documentation for the whole package. Remember that readers will often see it _without_ the accompanying code, so it needs to make sense on its own.
@@ -146,110 +119,6 @@ This is the _whole_ user manual for your code. It will be included in the autoge
 
 Any change to the `script` API should also be accompanied by an update to the README. If you add a new method, add it to the appropriate table (sources, filters, or sinks), and if it's the equivalent of a command Unix command, add it to the table of Unix equivalents too.
 
-# Writing pipe operations
-
-Here are some tips for writing sources, filters, and sinks for `script` pipes.
-
-## Writing a source
-
-All a pipe source has to do is return a pointer to a `script.Pipe`. To be useful, a pipe needs to have a reader (a data source, such as a file) associated with it.
-
-`Echo()` is a simple example that just creates a pipe containing a string:
-
-```go
-func Echo(s string) *script.Pipe {
-	return script.NewPipe().WithReader(strings.NewReader(s))
-}
-```
-
-Let's break this down:
-
-* We create a `strings.Reader` to be our data source, using `strings.NewReader` on the supplied string.
-* We create a new pipe with `NewPipe()`.
-* We attach the reader to the pipe with `WithReader()`.
-
-In fact, any `io.Reader` can be the data source for a pipe. Passing it to `WithReader()` will turn it into a `ReadAutoCloser`, which is a wrapper for `io.Reader` that automatically closes the reader once it has been fully read.
-
-Here's a possible implementation of `File()`, for example:
-
-```go
-func File(name string) *script.Pipe {
-	p := script.NewPipe()
-	f, err := os.Open(name)
-	if err != nil {
-		return p.WithError(err)
-	}
-	return p.WithReader(f)
-}
-```
-
-## Writing a filter
-
-Filters are methods on pipes, that return pipes. For example, here's a simple filter that just reads and rejects all input, returning an empty pipe:
-
-```go
-func (p *script.Pipe) RejectEverything() *script.Pipe {
-	if p == nil || p.Error() != nil {
-		return p
-	}
-	// don't care about result, just reading for effect
-	_, err := ioutil.ReadAll(p.Reader)
-	if err != nil {
-		p.SetError(err)
-		return p
-	}
-	return script.Echo("")
-}
-```
-
-Important things to note here:
-
-* The first thing we do is check the pipe's error status. If this is set, we do nothing, and just return the original pipe.
-* If an error occurs, we set the pipe's error status, using `p.SetError()`, and return the pipe.
-
-Filters must not log anything, terminate the program, or return anything but `*script.Pipe`.
-
-As you can see from the example, the pipe's reader is available to you as `p.Reader`. You can do anything with that that you can with an `io.Reader`.
-
-If your method modifies the pipe (for example if it can set an error on the pipe), it must take a pointer receiver, as in this example. For consistency, all methods should take a pointer receiver anyway, even if they don't modify the pipe (it's hard to imagine a useful method that wouldn't).
-
-Filters that are line-oriented can usually be written using `EachLine()` to simplify things. For example:
-
-```go
-func (p *Pipe) Match(s string) *Pipe {
-	return p.EachLine(func(line string, out *strings.Builder) {
-		if strings.Contains(line, s) {
-			out.WriteString(line)
-			out.WriteRune('\n')
-		}
-	})
-}
-```
-
-## Writing a sink
-
-Any method on a pipe that returns something other than a pipe is a sink. For example, here's an implementation of `String()`:
-
-```go
-func (p *script.Pipe) String() (string, error) {
-	if p.Error() != nil {
-		return "", p.Error()
-	}
-	res, err := ioutil.ReadAll(p.Reader)
-	if err != nil {
-		p.SetError(err)
-		return "", err
-	}
-	return string(res), nil
-}
-```
-
-Again, the first thing we do is check the error status on the pipe. If it's set, we return the zero value (empty string in this case) and the error.
-
-We then try to read the whole contents of the pipe. If we get an error on reading, we set the pipe's error status and return the zero value and the error.
-
-Otherwise, we return the result of reading the pipe, and a nil error.
-
 # Before submitting your pull request
 
 Here's a handy checklist for making sure your PR will be accepted as quickly as possible.
diff --git a/README.md b/README.md
index 2039f58..7333964 100644
--- a/README.md
+++ b/README.md
@@ -14,6 +14,9 @@ Why shouldn't it be as easy to write system administration programs in Go as it
 
 Shell scripts often compose a sequence of operations on a stream of data (a _pipeline_). This is how `script` works, too.
 
+> *This is one absolutely superb API design. Taking inspiration from shell pipes and turning it into a Go library with syntax this clean is really impressive.*\
+> —[Simon Willison](https://news.ycombinator.com/item?id=30649524)
+
 Read more: [Scripting with Go](https://bitfieldconsulting.com/golang/scripting)
 
 # Quick start: Unix equivalents
@@ -70,6 +73,12 @@ But what if, instead of reading a specific file, we want to simply pipe input in
 script.Stdin().Match("Error").Stdout()
 ```
 
+Just for fun, let's filter all the results through some arbitrary Go function:
+
+```go
+script.Stdin().Match("Error").FilterLine(strings.ToUpper).Stdout()
+```
+
 That was almost too easy! So let's pass in a list of files on the command line, and have our program read them all in sequence and output the matching lines:
 
 ```go
@@ -88,7 +97,48 @@ What's that? You want to append that output to a file instead of printing it to
 script.Args().Concat().Match("Error").First(10).AppendFile("/var/log/errors.txt")
 ```
 
-# Real use cases
+Suppose we want to execute some external program instead of doing the work ourselves. We can do that too:
+
+```go
+script.Exec("ping 127.0.0.1").Stdout()
+```
+
+But maybe we don't know the arguments yet; we might get them from the user, for example. We'd like to be able to run the external command repeatedly, each time passing it the next line of input. No worries:
+
+```go
+script.Args().ExecForEach("ping -c 1 {{.}}").Stdout()
+```
+
+If there isn't a built-in operation that does what we want, we can just write our own:
+
+```go
+script.Echo("hello world").Filter(func (r io.Reader, w io.Writer) error {
+	n, err := io.Copy(w, r)
+	fmt.Fprintf(w, "\nfiltered %d bytes\n", n)
+	return err
+}).Stdout()
+// Output:
+// hello world
+// filtered 11 bytes
+```
+
+Notice that the "hello world" appeared before the "filtered n bytes". Filters run concurrently, so the pipeline can start producing output before the input has been fully read.
+
+If we want to scan input line by line, we could do that with a `Filter` function that creates a `bufio.Scanner` on its input, but we don't need to:
+
+```go
+script.Echo("a\nb\nc").FilterScan(func(line string, w io.Writer) {
+	fmt.Fprintf(w, "scanned line: %q\n", line)
+}).Stdout()
+// Output:
+// scanned line: "a"
+// scanned line: "b"
+// scanned line: "c"
+```
+
+And there's more. Much more. [Read the docs](https://pkg.go.dev/github.com/bitfield/script) for full details, and more examples.
+
+# A realistic use case
 
 Let's use `script` to write a program that system administrators might actually need. One thing I often find myself doing is counting the most frequent visitors to a website over a given period of time. Given an Apache log in the Common Log Format like this:
 
@@ -158,22 +208,26 @@ Filters are methods on an existing pipe that also return a pipe, allowing you to
 | [`Column`](https://pkg.go.dev/github.com/bitfield/script#Pipe.Column) | Nth column of input |
 | [`Concat`](https://pkg.go.dev/github.com/bitfield/script#Pipe.Concat) | contents of multiple files |
 | [`Dirname`](https://pkg.go.dev/github.com/bitfield/script#Pipe.Dirname) | removes filename from each line, leaving only leading path components |
-| [`EachLine`](https://pkg.go.dev/github.com/bitfield/script#Pipe.EachLine) | user-supplied function |
 | [`Echo`](https://pkg.go.dev/github.com/bitfield/script#Pipe.Echo) | all input replaced by given string |
 | [`Exec`](https://pkg.go.dev/github.com/bitfield/script#Pipe.Exec) | filtered through external command |
 | [`ExecForEach`](https://pkg.go.dev/github.com/bitfield/script#Pipe.ExecForEach) | execute given command template for each line of input |
-| [`First`](https://pkg.go.dev/github.com/bitfield/script#Pipe.First) | first N lines |
+| [`Filter`](https://pkg.go.dev/github.com/bitfield/script#Pipe.Filter) | user-supplied function filtering a reader to a writer |
+| [`FilterLine`](https://pkg.go.dev/github.com/bitfield/script#Pipe.FilterLine) | user-supplied function filtering each line to a string|
+| [`FilterScan`](https://pkg.go.dev/github.com/bitfield/script#Pipe.FilterScan) | user-supplied function filtering each line to a writer |
+| [`First`](https://pkg.go.dev/github.com/bitfield/script#Pipe.First) | first N lines of input |
 | [`Freq`](https://pkg.go.dev/github.com/bitfield/script#Pipe.Freq) | frequency count of unique input lines, most frequent first |
 | [`Join`](https://pkg.go.dev/github.com/bitfield/script#Pipe.Join) | replace all newlines with spaces |
-| [`Last`](https://pkg.go.dev/github.com/bitfield/script#Pipe.Last) | last N lines |
-| [`Match`](https://pkg.go.dev/github.com/bitfield/script#Pipe.Match) | matching lines |
-| [`MatchRegexp`](https://pkg.go.dev/github.com/bitfield/script#Pipe.MatchRegexp) | matching lines |
-| [`Reject`](https://pkg.go.dev/github.com/bitfield/script#Pipe.Reject) | non-matching lines |
-| [`RejectRegexp`](https://pkg.go.dev/github.com/bitfield/script#Pipe.RejectRegexp) | non-matching lines |
-| [`Replace`](https://pkg.go.dev/github.com/bitfield/script#Pipe.Replace) | matching lines replaced with string |
-| [`ReplaceRegexp`](https://pkg.go.dev/github.com/bitfield/script#Pipe.ReplaceRegexp) | matching lines replaced with string |
+| [`Last`](https://pkg.go.dev/github.com/bitfield/script#Pipe.Last) | last N lines of input|
+| [`Match`](https://pkg.go.dev/github.com/bitfield/script#Pipe.Match) | lines matching given string |
+| [`MatchRegexp`](https://pkg.go.dev/github.com/bitfield/script#Pipe.MatchRegexp) | lines matching given regexp |
+| [`Reject`](https://pkg.go.dev/github.com/bitfield/script#Pipe.Reject) | lines not matching given string |
+| [`RejectRegexp`](https://pkg.go.dev/github.com/bitfield/script#Pipe.RejectRegexp) | lines not matching given regexp |
+| [`Replace`](https://pkg.go.dev/github.com/bitfield/script#Pipe.Replace) | matching text replaced with given string |
+| [`ReplaceRegexp`](https://pkg.go.dev/github.com/bitfield/script#Pipe.ReplaceRegexp) | matching text replaced with given string |
 | [`SHA256Sums`](https://pkg.go.dev/github.com/bitfield/script#Pipe.SHA256Sums) | SHA-256 hashes of each listed file |
 
+Note that filters run concurrently, rather than producing nothing until each stage has fully read its input. This is convenient for executing long-running comands, for example. If you do need to wait for the pipeline to complete, call `Wait`.
+
 ## Sinks
 
 Sinks are methods that return some data from a pipe, ending the pipeline and extracting its full contents in a specified way:
@@ -188,12 +242,9 @@ Sinks are methods that return some data from a pipe, ending the pipeline and ext
 | [`Slice`](https://pkg.go.dev/github.com/bitfield/script#Pipe.Slice) | | data as `[]string`, error  |
 | [`Stdout`](https://pkg.go.dev/github.com/bitfield/script#Pipe.Stdout) | standard output | bytes written, error  |
 | [`String`](https://pkg.go.dev/github.com/bitfield/script#Pipe.String) | | data as `string`, error  |
+| [`Wait`](https://pkg.go.dev/github.com/bitfield/script#Pipe.Wait) | | none  |
 | [`WriteFile`](https://pkg.go.dev/github.com/bitfield/script#Pipe.WriteFile) | specified file, truncating if it exists | bytes written, error  |
 
-## Examples
-
-Since `script` is designed to help you write system administration programs, a few simple examples of such programs are included in the [examples](examples/) directory.
-
 # Contributing
 
 See the [contributor's guide](CONTRIBUTING.md) for some helpful tips if you'd like to contribute to the `script` project.
diff --git a/examples/cat/main.go b/examples/cat/main.go
deleted file mode 100644
index d1c3ed5..0000000
--- a/examples/cat/main.go
+++ /dev/null
@@ -1,9 +0,0 @@
-package main
-
-import "github.com/bitfield/script"
-
-// The filter form of cat copies its standard input to its standard output.
-
-func main() {
-	script.Stdin().Stdout()
-}
diff --git a/examples/cat2/main.go b/examples/cat2/main.go
deleted file mode 100644
index d74a24a..0000000
--- a/examples/cat2/main.go
+++ /dev/null
@@ -1,12 +0,0 @@
-package main
-
-import (
-	"github.com/bitfield/script"
-)
-
-// This version of cat takes a list of files on the command line, and prints
-// their concatenated contents.
-
-func main() {
-	script.Args().Concat().Stdout()
-}
diff --git a/examples/echo/main.go b/examples/echo/main.go
deleted file mode 100644
index 0bf449a..0000000
--- a/examples/echo/main.go
+++ /dev/null
@@ -1,7 +0,0 @@
-package main
-
-import "github.com/bitfield/script"
-
-func main() {
-	script.Args().Join().Stdout()
-}
diff --git a/examples/execforeach/main.go b/examples/execforeach/main.go
deleted file mode 100644
index 22e5ccf..0000000
--- a/examples/execforeach/main.go
+++ /dev/null
@@ -1,11 +0,0 @@
-package main
-
-import (
-	"github.com/bitfield/script"
-)
-
-// This program prints out the names of all files in the current directory.
-
-func main() {
-	script.ListFiles("*").ExecForEach("echo {{.}}").Stdout()
-}
diff --git a/examples/execute/main.go b/examples/execute/main.go
deleted file mode 100644
index 73a947a..0000000
--- a/examples/execute/main.go
+++ /dev/null
@@ -1,9 +0,0 @@
-package main
-
-import (
-	"github.com/bitfield/script"
-)
-
-func main() {
-	script.Exec("bash -c 'echo hello world'").Stdout()
-}
diff --git a/examples/find/main.go b/examples/find/main.go
deleted file mode 100644
index cfc2b32..0000000
--- a/examples/find/main.go
+++ /dev/null
@@ -1,15 +0,0 @@
-package main
-
-import (
-	"os"
-
-	"github.com/bitfield/script"
-)
-
-func main() {
-	listPath := "."
-	if len(os.Args) > 1 {
-		listPath = os.Args[1]
-	}
-	script.FindFiles(listPath).Stdout()
-}
diff --git a/examples/grep/main.go b/examples/grep/main.go
deleted file mode 100644
index 5a686ca..0000000
--- a/examples/grep/main.go
+++ /dev/null
@@ -1,11 +0,0 @@
-package main
-
-import (
-	"os"
-
-	"github.com/bitfield/script"
-)
-
-func main() {
-	script.Stdin().Match(os.Args[1]).Stdout()
-}
diff --git a/examples/head/main.go b/examples/head/main.go
deleted file mode 100644
index bf15466..0000000
--- a/examples/head/main.go
+++ /dev/null
@@ -1,17 +0,0 @@
-package main
-
-import (
-	"log"
-	"os"
-	"strconv"
-
-	"github.com/bitfield/script"
-)
-
-func main() {
-	lines, err := strconv.Atoi(os.Args[1])
-	if err != nil {
-		log.Fatal(err)
-	}
-	script.Stdin().First(lines).Stdout()
-}
diff --git a/examples/least_freq/main.go b/examples/least_freq/main.go
deleted file mode 100644
index 7d2a174..0000000
--- a/examples/least_freq/main.go
+++ /dev/null
@@ -1,9 +0,0 @@
-package main
-
-import "github.com/bitfield/script"
-
-// filters the least frequent line from input and copies it to standard output
-
-func main() {
-	script.Stdin().Freq().Last(1).Stdout()
-}
diff --git a/examples/ls/main.go b/examples/ls/main.go
deleted file mode 100644
index 69b0aec..0000000
--- a/examples/ls/main.go
+++ /dev/null
@@ -1,19 +0,0 @@
-package main
-
-import (
-	"os"
-	"regexp"
-
-	"github.com/bitfield/script"
-)
-
-// Like Unix `ls`, hide files starting with '.'
-var hideDotFiles = regexp.MustCompile(`[^/]*/\\.[^/]*$`)
-
-func main() {
-	listPath := "."
-	if len(os.Args) > 1 {
-		listPath = os.Args[1]
-	}
-	script.ListFiles(listPath).RejectRegexp(hideDotFiles).Stdout()
-}
diff --git a/examples/sha256sum/main.go b/examples/sha256sum/main.go
deleted file mode 100644
index 7a47eb5..0000000
--- a/examples/sha256sum/main.go
+++ /dev/null
@@ -1,27 +0,0 @@
-package main
-
-import (
-	"fmt"
-	"github.com/bitfield/script"
-	"log"
-	"os"
-	"strings"
-)
-
-func main() {
-	// Read the first 2 args, and calculate the checksum of the files.
-	hashFiles, err := script.Args().First(2).SHA256Sums().String()
-	if err != nil {
-		log.Fatal(err)
-	}
-
-	// Compare the SHA256 checksum to check if files are identical.
-	checkSums := strings.Split(hashFiles,"\n")
-	if checkSums[0] == checkSums[1] {
-		fmt.Println("Hashes are identical")
-		os.Exit(0)
-	}
-
-	fmt.Printf("Hashes are different: %q vs %q\n", checkSums[0], checkSums[1])
-	os.Exit(1)
-}
diff --git a/examples/slice/errors.txt b/examples/slice/errors.txt
deleted file mode 100644
index f64525e..0000000
--- a/examples/slice/errors.txt
+++ /dev/null
@@ -1,13 +0,0 @@
-Line containing the word 'error'
-Another line containing an error message
-Oh no, an error occurred!
-Here's another error
-More errors
-To err is human, but forgiveness is surprisingly difficult.
-something something error happened
-error: a suffusion of yellow
-An UNKNOWN ERROR occurred
-Drive Z: does not exist, please try again.
-error deactivating tyrannosaur fence: you didn't say the magic word!
-error: the operation completed successfully
-Fatal error: keyboard not present. Press F1 to continue.
\ No newline at end of file
diff --git a/examples/slice/main.go b/examples/slice/main.go
deleted file mode 100644
index d6868da..0000000
--- a/examples/slice/main.go
+++ /dev/null
@@ -1,27 +0,0 @@
-package main
-
-import (
-	"fmt"
-	"log"
-
-	"github.com/bitfield/script"
-)
-
-// This program reads all the files supplied to it on the command line, and
-// prints out the first ten lines matching the string 'error', with an 'ERROR:'
-// prefix.
-//
-// For example:
-// go run main.go errors.txt
-// ERROR: Line containing the word 'error'
-// ...
-
-func main() {
-	errors, err := script.Args().Concat().Match("error").First(10).Slice()
-	if err != nil {
-		log.Fatal(err)
-	}
-	for _, e := range errors {
-		fmt.Printf("ERROR: %s\n", e)
-	}
-}
diff --git a/examples/tail/main.go b/examples/tail/main.go
deleted file mode 100644
index 50d4645..0000000
--- a/examples/tail/main.go
+++ /dev/null
@@ -1,17 +0,0 @@
-package main
-
-import (
-	"log"
-	"os"
-	"strconv"
-
-	"github.com/bitfield/script"
-)
-
-func main() {
-	lines, err := strconv.Atoi(os.Args[1])
-	if err != nil {
-		log.Fatal(err)
-	}
-	script.Stdin().Last(lines).Stdout()
-}
diff --git a/examples/todo/examples.text b/examples/todo/examples.text
deleted file mode 100644
index 5e1fdbd..0000000
--- a/examples/todo/examples.text
+++ /dev/null
@@ -1,94 +0,0 @@
-package script
-
-import (
-	"bufio"
-	"fmt"
-	"os"
-	"regexp"
-)
-
-type Todo struct {
-	LineNumber int32
-	Message    string
-	FilePath   string
-}
-/*** 
-todo 1
-todo 2
-todo 3
-*/
-
-/** todo ddjdjdj */
-
-# todo say something
-// todo hello world
-// FindTodos Open file and find todos
-func FindTodos(filePath string) (*[]Todo, error) {
-
-	file, err := os.Open(filePath)
-
-	defer file.Close()
-
-	if err != nil {
-		return nil, fmt.Errorf("Could not open file %s: %v", filePath, err)
-	}
-
-	scanner := bufio.NewScanner(file)
-	scanner.Split(bufio.ScanLines)
-
-	lineNumber := 1
-
-	foundTodos := make([]Todo, 0)
-
-	for scanner.Scan() {
-		// find todos in every line
-		// ideally, the todo will be inside a comment bloc i.e. // /**/ ///
-		findTodo, err := regexp.Match(`((?i)\/\/|#).todo.*`, scanner.Bytes()) ///// todo another todo here
-
-		if err != nil {
-			return nil, fmt.Errorf("An error occurred while finding todo: %v", err)
-		}
-
-		if findTodo == true {
-
-			todo := Todo{
-				LineNumber: int32(lineNumber),
-				Message:    scanner.Text(),
-				FilePath:   filePath,
-			}
-
-			foundTodos = append(foundTodos, todo)
-
-			fmt.Printf("%d: %s\n", lineNumber, scanner.Text())
-
-		}
-
-		lineNumber++
-	}
-
-	return &foundTodos, nil
-}
-
-// Opts Options for getting files in a directory/subdirectory
-type Opts struct {
-	RespectGitignore bool
-	SkipHiddenFiles  bool
-}
-
-// GetFilesInDirectory scan directory and subdirectories, returning all files
-// except those that are exempted. Exempted files could be either by an exact match or
-// a regex match. Directories can also be exempted.
-func GetFilesInDirectory(opts Opts, exempt ...string) ([]string, error) {
-	// TODO read all files and directories inside
-	// TODO have a list of files to exempt, either by
-	// todos: extension, filename or regex pattern
-	// todo: for each file, check if there are any todos
-
-	return nil, nil
-
-}
-
-// GetGitingoreFiles find all gitingore files and return files to ignore
-func GetGitingoreFiles() ([]string, error) {
-	return nil, nil
-}
diff --git a/examples/todo/main.go b/examples/todo/main.go
deleted file mode 100644
index d005975..0000000
--- a/examples/todo/main.go
+++ /dev/null
@@ -1,31 +0,0 @@
-package main
-
-import (
-	"fmt"
-	"os"
-	"regexp"
-	"strings"
-
-	"github.com/bitfield/script"
-)
-
-func main() {
-	listPath := "."
-	if len(os.Args) > 1 {
-		listPath = os.Args[1]
-	}
-	// filter hidden directories and files
-	filterFiles := regexp.MustCompile(`^\..*|/\.`)
-	files := script.FindFiles(listPath).RejectRegexp(filterFiles)
-	content := files.EachLine(func(filePath string, builderFile *strings.Builder) {
-		p := script.File(filePath)
-		lineNumber := 1
-		p.EachLine(func(str string, build *strings.Builder) {
-			if strings.Contains(str, "todo") {
-				builderFile.WriteString(fmt.Sprintf("%s:%d %s \n", filePath, lineNumber, strings.TrimSpace(str)))
-			}
-			lineNumber++
-		})
-	})
-	content.Stdout()
-}
diff --git a/examples/visitors/access.log b/examples/visitors/access.log
deleted file mode 100644
index 5049388..0000000
--- a/examples/visitors/access.log
+++ /dev/null
@@ -1,25 +0,0 @@
-212.205.21.11 - - [30/Jun/2019:17:06:15 +0000] "GET / HTTP/1.1" 200 2028 "https://example.com/ "Mozilla/5.0 (Linux; Android 8.0.0; FIG-LX1 Build/HUAWEIFIG-LX1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/64.0.3282.156 Mobile Safari/537.36"
-212.205.21.11 - - [30/Jun/2019:17:06:15 +0000] "GET / HTTP/1.1" 200 162544 "https://example.com/ "Mozilla/5.0 (Linux; Android 8.0.0; FIG-LX1 Build/HUAWEIFIG-LX1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/64.0.3282.156 Mobile Safari/537.36"
-212.205.21.11 - - [30/Jun/2019:17:06:15 +0000] "GET / HTTP/1.1" 200 9419 "https://example.com/ "Mozilla/5.0 (Linux; Android 8.0.0; FIG-LX1 Build/HUAWEIFIG-LX1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/64.0.3282.156 Mobile Safari/537.36"
-212.205.21.11 - - [30/Jun/2019:17:06:15 +0000] "GET / HTTP/1.1" 200 2058 "https://example.com/ "Mozilla/5.0 (Linux; Android 8.0.0; FIG-LX1 Build/HUAWEIFIG-LX1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/64.0.3282.156 Mobile Safari/537.36"
-212.205.21.11 - - [30/Jun/2019:17:06:15 +0000] "GET / HTTP/1.1" 200 343743 "https://example.com/ "Mozilla/5.0 (Linux; Android 8.0.0; FIG-LX1 Build/HUAWEIFIG-LX1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/64.0.3282.156 Mobile Safari/537.36"
-212.205.21.11 - - [30/Jun/2019:17:06:16 +0000] "GET / HTTP/1.1" 200 1150 "https://example.com/ "Mozilla/5.0 (Linux; Android 8.0.0; FIG-LX1 Build/HUAWEIFIG-LX1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/64.0.3282.156 Mobile Safari/537.36"
-212.205.21.11 - - [30/Jun/2019:17:06:16 +0000] "GET / HTTP/1.1" 200 2946 "https://example.com/ "Mozilla/5.0 (Linux; Android 8.0.0; FIG-LX1 Build/HUAWEIFIG-LX1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/64.0.3282.156 Mobile Safari/537.36"
-176.182.2.191 - - [30/Jun/2019:17:06:17 +0000] "GET / HTTP/1.1" 200 13278 "https://example.com/ "Mozilla/5.0 (iPhone; CPU iPhone OS 12_3_1 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/12.1.1 Mobile/15E148 Safari/604.1"
-176.182.2.191 - - [30/Jun/2019:17:06:19 +0000] "GET / HTTP/1.1" 200 29474 "https://example.com/ "Mozilla/5.0 (iPhone; CPU iPhone OS 12_3_1 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/12.1.1 Mobile/15E148 Safari/604.1"
-176.182.2.191 - - [30/Jun/2019:17:06:19 +0000] "GET / HTTP/1.1" 200 29349 "https://example.com/ "Mozilla/5.0 (iPhone; CPU iPhone OS 12_3_1 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/12.1.1 Mobile/15E148 Safari/604.1"
-176.182.2.191 - - [30/Jun/2019:17:06:19 +0000] "GET / HTTP/1.1" 200 48271 "https://example.com/ "Mozilla/5.0 (iPhone; CPU iPhone OS 12_3_1 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/12.1.1 Mobile/15E148 Safari/604.1"
-176.182.2.191 - - [30/Jun/2019:17:06:19 +0000] "GET / HTTP/1.1" 200 1380 "https://example.com/ "Mozilla/5.0 (iPhone; CPU iPhone OS 12_3_1 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/12.1.1 Mobile/15E148 Safari/604.1"
-176.182.2.191 - - [30/Jun/2019:17:06:20 +0000] "GET / HTTP/1.1" 200 2028 "https://example.com/ "Mozilla/5.0 (iPhone; CPU iPhone OS 12_3_1 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/12.1.1 Mobile/15E148 Safari/604.1"
-176.182.2.191 - - [30/Jun/2019:17:06:19 +0000] "GET / HTTP/1.1" 200 91819 "https://example.com/ "Mozilla/5.0 (iPhone; CPU iPhone OS 12_3_1 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/12.1.1 Mobile/15E148 Safari/604.1"
-176.182.2.191 - - [30/Jun/2019:17:06:19 +0000] "GET / HTTP/1.1" 200 305667 "https://example.com/ "Mozilla/5.0 (iPhone; CPU iPhone OS 12_3_1 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/12.1.1 Mobile/15E148 Safari/604.1"
-176.182.2.191 - - [30/Jun/2019:17:06:20 +0000] "GET / HTTP/1.1" 200 13194 "https://example.com/ "Mozilla/5.0 (iPhone; CPU iPhone OS 12_3_1 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/12.1.1 Mobile/15E148 Safari/604.1"
-176.182.2.191 - - [30/Jun/2019:17:06:20 +0000] "GET / HTTP/1.1" 200 12935 "https://example.com/ "Mozilla/5.0 (iPhone; CPU iPhone OS 12_3_1 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/12.1.1 Mobile/15E148 Safari/604.1"
-176.182.2.191 - - [30/Jun/2019:17:06:20 +0000] "GET / HTTP/1.1" 200 14598 "https://example.com/ "Mozilla/5.0 (iPhone; CPU iPhone OS 12_3_1 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/12.1.1 Mobile/15E148 Safari/604.1"
-176.182.2.191 - - [30/Jun/2019:17:06:20 +0000] "GET / HTTP/1.1" 200 22458 "https://example.com/ "Mozilla/5.0 (iPhone; CPU iPhone OS 12_3_1 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/12.1.1 Mobile/15E148 Safari/604.1"
-176.182.2.191 - - [30/Jun/2019:17:06:20 +0000] "GET / HTTP/1.1" 200 15737 "https://example.com/ "Mozilla/5.0 (iPhone; CPU iPhone OS 12_3_1 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/12.1.1 Mobile/15E148 Safari/604.1"
-176.182.2.191 - - [30/Jun/2019:17:06:20 +0000] "GET / HTTP/1.1" 404 17679 "https://example.com/ "Mozilla/5.0 (iPhone; CPU iPhone OS 12_3_1 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/12.1.1 Mobile/15E148 Safari/604.1"
-176.182.2.191 - - [30/Jun/2019:17:06:23 +0000] "GET / HTTP/1.1" 200 5995 "https://example.com/ "Mozilla/5.0 (iPhone; CPU iPhone OS 12_3_1 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/12.1.1 Mobile/15E148 Safari/604.1"
-190.253.121.1 - - [30/Jun/2019:17:06:23 +0000] "GET / HTTP/1.1" 200 8809 "-" "Mozilla/5.0 (Linux; Android 9; SAMSUNG SM-J415FN Build/PPR1.180610.011) AppleWebKit/537.36 (KHTML, like Gecko) SamsungBrowser/9.2 Chrome/67.0.3396.87 Mobile Safari/537.36"
-176.182.2.191 - - [30/Jun/2019:17:06:24 +0000] "GET / HTTP/1.1" 200 162544 "https://example.com/ "Mozilla/5.0 (iPhone; CPU iPhone OS 12_3_1 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/12.1.1 Mobile/15E148 Safari/604.1"
-90.53.111.17 - - [30/Jun/2019:17:06:24 +0000] "GET / HTTP/1.1" 200 - "https://example.com/ "Mozilla/5.0 (Linux; Android 9; SAMSUNG SM-J415FN Build/PPR1.180610.011) AppleWebKit/537.36 (KHTML, like Gecko) SamsungBrowser/9.2 Chrome/67.0.3396.87 Mobile Safari/537.36"
\ No newline at end of file
diff --git a/examples/visitors/main.go b/examples/visitors/main.go
deleted file mode 100644
index 767c464..0000000
--- a/examples/visitors/main.go
+++ /dev/null
@@ -1,24 +0,0 @@
-/*
-This program reads an Apache logfile in Common Log Format, like this:
-
-212.205.21.11 - - [30/Jun/2019:17:06:15 +0000] "GET / HTTP/1.1" 200 2028 "https://example.com/ "Mozilla/5.0 (Linux; Android 8.0.0; FIG-LX1 Build/HUAWEIFIG-LX1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/64.0.3282.156 Mobile Safari/537.36"
-
-It extracts the first column of each line (the visitor IP address), counts the
-frequency of each unique IP address in the log, and outputs the 10 most frequent
-visitors in the log. Example output:
-
-16 176.182.2.191
-7 212.205.21.11
-1 190.253.121.1
-1 90.53.111.17
-
-*/
-package main
-
-import (
-	"github.com/bitfield/script"
-)
-
-func main() {
-	script.Stdin().Column(1).Freq().First(10).Stdout()
-}
diff --git a/go.mod b/go.mod
index e9cc35c..813e5aa 100644
--- a/go.mod
+++ b/go.mod
@@ -4,5 +4,6 @@ go 1.12
 
 require (
 	bitbucket.org/creachadair/shell v0.0.7
-	github.com/google/go-cmp v0.5.6
+	github.com/google/go-cmp v0.5.7
+	golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 // indirect
 )
diff --git a/go.sum b/go.sum
index 214de67..393c9f0 100644
--- a/go.sum
+++ b/go.sum
@@ -1,6 +1,8 @@
 bitbucket.org/creachadair/shell v0.0.7 h1:Z96pB6DkSb7F3Y3BBnJeOZH2gazyMTWlvecSD4vDqfk=
 bitbucket.org/creachadair/shell v0.0.7/go.mod h1:oqtXSSvSYr4624lnnabXHaBsYW6RD80caLi2b3hJk0U=
-github.com/google/go-cmp v0.5.6 h1:BKbKCqvP6I+rmFHt06ZmyQtvB8xAkWdhFyr0ZUNZcxQ=
 github.com/google/go-cmp v0.5.6/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
-golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4=
+github.com/google/go-cmp v0.5.7 h1:81/ik6ipDQS2aGcBfIN5dHDB36BwrStyeAQquSYCV4o=
+github.com/google/go-cmp v0.5.7/go.mod h1:n+brtR0CgQNWTVd5ZUFpTBC8YFBDLK/h/bpaJ8/DtOE=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 h1:go1bK/D/BFZV2I8cIQd1NKEZ+0owSTG1fDTci4IqFcE=
+golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
diff --git a/script.go b/script.go
index 6837df8..fe0c274 100644
--- a/script.go
+++ b/script.go
@@ -2,14 +2,11 @@ package script
 
 import (
 	"bufio"
-	"bytes"
 	"container/ring"
 	"crypto/sha256"
 	"encoding/hex"
-	"errors"
 	"fmt"
 	"io"
-	"io/ioutil"
 	"os"
 	"os/exec"
 	"path/filepath"
@@ -17,6 +14,7 @@ import (
 	"sort"
 	"strconv"
 	"strings"
+	"sync"
 	"text/template"
 
 	"bitbucket.org/creachadair/shell"
@@ -56,7 +54,7 @@ func (a ReadAutoCloser) Close() error {
 // closable.
 func NewReadAutoCloser(r io.Reader) ReadAutoCloser {
 	if _, ok := r.(io.Closer); !ok {
-		return ReadAutoCloser{ioutil.NopCloser(r)}
+		return ReadAutoCloser{io.NopCloser(r)}
 	}
 	rc, ok := r.(io.ReadCloser)
 	if !ok {
@@ -69,34 +67,36 @@ func NewReadAutoCloser(r io.Reader) ReadAutoCloser {
 // Pipe represents a pipe object with an associated ReadAutoCloser.
 type Pipe struct {
 	Reader ReadAutoCloser
-	err    error
 	stdout io.Writer
+
+	// because pipe stages are concurrent, protect 'err'
+	mu  *sync.Mutex
+	err error
 }
 
 // NewPipe returns a pointer to a new empty pipe.
 func NewPipe() *Pipe {
 	return &Pipe{
 		Reader: ReadAutoCloser{},
+		mu:     &sync.Mutex{},
 		err:    nil,
 		stdout: os.Stdout,
 	}
 }
 
-// Close closes the pipe's associated reader. This is always safe to do, because
-// pipes created from a non-closable source will have an `ioutil.NopCloser` to
-// call.
+// Close closes the pipe's associated reader. This is a no-op if the reader is
+// not also a Closer.
 func (p *Pipe) Close() error {
-	if p == nil {
-		return nil
-	}
 	return p.Reader.Close()
 }
 
-// Error returns the last error returned by any pipe operation, or nil otherwise.
+// Error returns any error present on the pipe, or nil otherwise.
 func (p *Pipe) Error() error {
-	if p == nil {
+	if p.mu == nil { // uninitialised pipe
 		return nil
 	}
+	p.mu.Lock()
+	defer p.mu.Unlock()
 	return p.err
 }
 
@@ -128,46 +128,37 @@ func (p *Pipe) ExitStatus() int {
 // Unlike most sinks, Read does not necessarily read the whole contents of the
 // pipe. It will read as many bytes as it takes to fill the slice.
 func (p *Pipe) Read(b []byte) (int, error) {
-	if p == nil {
-		return 0, io.EOF
-	}
 	return p.Reader.Read(b)
 }
 
-// SetError sets the pipe's error status to the specified error.
+// SetError sets the specified error on the pipe.
 func (p *Pipe) SetError(err error) {
-	if p != nil {
-		if err != nil {
-			p.Close()
-		}
-		p.err = err
+	if p.mu == nil { // uninitialised pipe
+		return
+	}
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	if err != nil {
+		p.Close()
 	}
+	p.err = err
 }
 
-// WithReader takes an io.Reader, and associates the pipe with that reader. If
-// necessary, the reader will be automatically closed once it has been
-// completely read.
+// WithReader sets the pipe's input to the specified reader. If necessary, the
+// reader will be automatically closed once it has been completely read.
 func (p *Pipe) WithReader(r io.Reader) *Pipe {
-	if p == nil {
-		return nil
-	}
 	p.Reader = NewReadAutoCloser(r)
 	return p
 }
 
-// WithStdout takes an io.Writer, and associates the pipe's standard output with
-// that reader, instead of the default os.Stdout. This is primarily useful for
-// testing.
+// WithStdout sets the pipe's standard output to the specified reader, instead
+// of the default os.Stdout.
 func (p *Pipe) WithStdout(w io.Writer) *Pipe {
-	if p == nil {
-		return nil
-	}
 	p.stdout = w
 	return p
 }
 
-// WithError sets the pipe's error status to the specified error and returns the
-// modified pipe.
+// WithError sets the specified error on the pipe and returns the modified pipe.
 func (p *Pipe) WithError(err error) *Pipe {
 	p.SetError(err)
 	return p
@@ -183,12 +174,12 @@ func Args() *Pipe {
 	return Echo(s.String())
 }
 
-// Echo returns a pipe containing the supplied string.
+// Echo creates a pipe containing the supplied string.
 func Echo(s string) *Pipe {
 	return NewPipe().WithReader(strings.NewReader(s))
 }
 
-// Exec runs an external command and returns a pipe containing its combined
+// Exec runs an external command and creates a pipe containing its combined
 // output (stdout and stderr).
 //
 // If the command had a non-zero exit status, the pipe's error status will also
@@ -208,9 +199,7 @@ func Exec(s string) *Pipe {
 	return NewPipe().Exec(s)
 }
 
-// File returns a *Pipe that reads from the specified file. This is useful for
-// starting pipelines. If there is an error opening the file, the pipe's error
-// status will be set.
+// File creates a pipe that reads from the file at the specified path.
 func File(name string) *Pipe {
 	p := NewPipe()
 	f, err := os.Open(name)
@@ -220,7 +209,7 @@ func File(name string) *Pipe {
 	return p.WithReader(f)
 }
 
-// FindFiles takes a directory path and returns a pipe listing all the files in
+// FindFiles takes a directory path and creates a pipe listing all the files in
 // the directory and its subdirectories recursively, one per line, like Unix
 // `find -type f`. If the path doesn't exist or can't be read, the pipe's error
 // status will be set.
@@ -248,18 +237,19 @@ func FindFiles(path string) *Pipe {
 	return Slice(fileNames)
 }
 
-// IfExists tests whether the specified file exists, and returns a pipe whose
+// IfExists tests whether the specified file exists, and creates a pipe whose
 // error status reflects the result. If the file doesn't exist, the pipe's error
 // status will be set, and if the file does exist, the pipe will have no error
 // status. This can be used to do some operation only if a given file exists:
 //
 // IfExists("/foo/bar").Exec("/usr/bin/something")
 func IfExists(filename string) *Pipe {
+	p := NewPipe()
 	_, err := os.Stat(filename)
 	if err != nil {
-		return NewPipe().WithError(err)
+		return p.WithError(err)
 	}
-	return NewPipe()
+	return p
 }
 
 // ListFiles creates a pipe containing the files and directories matching the
@@ -277,7 +267,7 @@ func ListFiles(path string) *Pipe {
 		}
 		return Slice(fileNames)
 	}
-	files, err := ioutil.ReadDir(path)
+	files, err := os.ReadDir(path)
 	if err != nil {
 		// Check for the case where the path matches exactly one file
 		s, err := os.Stat(path)
@@ -296,13 +286,13 @@ func ListFiles(path string) *Pipe {
 	return Slice(fileNames)
 }
 
-// Slice returns a pipe containing each element of the supplied slice of
+// Slice creates a pipe containing each element of the supplied slice of
 // strings, one per line.
 func Slice(s []string) *Pipe {
 	return Echo(strings.Join(s, "\n") + "\n")
 }
 
-// Stdin returns a pipe that reads from the program's standard input.
+// Stdin creates a pipe that reads from os.Stdin.
 func Stdin() *Pipe {
 	return NewPipe().WithReader(os.Stdin)
 }
@@ -315,34 +305,25 @@ func Stdin() *Pipe {
 // If a line is empty, Basename will produce '.'. Trailing slashes are removed.
 // The behaviour of Basename is the same as filepath.Base (not by coincidence).
 func (p *Pipe) Basename() *Pipe {
-	if p == nil || p.Error() != nil {
-		return p
-	}
-	return p.EachLine(func(line string, out *strings.Builder) {
-		out.WriteString(filepath.Base(line))
-		out.WriteRune('\n')
-	})
+	return p.FilterLine(filepath.Base)
 }
 
-// Column reads from the pipe, and returns a new pipe containing only the Nth
-// column of each line in the input, where '1' means the first column, and
-// columns are delimited by whitespace. Specifically, whatever Unicode defines
-// as whitespace ('WSpace=yes'). If there is an error reading the pipe, the
-// pipe's error status is also set.
+// Column produces only the Nth column of each line of input, where '1' is the
+// first column, and columns are delimited by whitespace. Specifically, whatever
+// Unicode defines as whitespace ('WSpace=yes').
 //
-// Lines containing less than N columns will be ignored.
+// Lines containing less than N columns will be dropped altogether.
 func (p *Pipe) Column(col int) *Pipe {
-	return p.EachLine(func(line string, out *strings.Builder) {
+	return p.FilterScan(func(line string, w io.Writer) {
 		columns := strings.Fields(line)
 		if col > 0 && col <= len(columns) {
-			out.WriteString(columns[col-1])
-			out.WriteRune('\n')
+			fmt.Fprintln(w, columns[col-1])
 		}
 	})
 }
 
-// Concat reads a list of filenames from the pipe, one per line, and returns a
-// pipe that reads all those files in sequence. If there are any errors (for
+// Concat reads a list of file paths from the pipe, one per line, and produces
+// the contents of all those files in sequence. If there are any errors (for
 // example, non-existent files), these will be ignored, execution will continue,
 // and the pipe's error status will not be set.
 //
@@ -364,169 +345,191 @@ func (p *Pipe) Column(col int) *Pipe {
 // without setting the pipe's error status. This mimics the behaviour of Unix
 // `cat`.
 func (p *Pipe) Concat() *Pipe {
-	if p == nil || p.Error() != nil {
-		return p
-	}
 	var readers []io.Reader
-	scanner := bufio.NewScanner(p.Reader)
-	for scanner.Scan() {
-		input, err := os.Open(scanner.Text())
+	p.FilterScan(func(line string, w io.Writer) {
+		input, err := os.Open(line)
 		if err != nil {
-			continue // skip errors
+			return // skip errors
 		}
 		readers = append(readers, NewReadAutoCloser(input))
-	}
-	err := scanner.Err()
-	if err != nil {
-		p.SetError(err)
-	}
+	}).Wait()
 	return p.WithReader(io.MultiReader(readers...))
 }
 
-// Dirname reads a list of pathnames from the pipe, one per line, and returns a
-// pipe that contains only the parent directories of each pathname. For example,
+// Dirname reads a list of pathnames from the pipe, one per line, and produces
+// only the parent directories of each pathname. For example,
 // `/usr/local/bin/foo` would become just `/usr/local/bin`. This is the
 // complementary operation to Basename.
 //
 // If a line is empty, Dirname will produce a '.'. Trailing slashes are removed,
-// unless Dirname returns the root folder. The behaviour of Dirname is the same
-// as filepath.Dir (not by coincidence).
+// unless Dirname returns the root folder. Otherwise, the behaviour of Dirname
+// is the same as filepath.Dir (not by coincidence).
 func (p *Pipe) Dirname() *Pipe {
-	if p == nil || p.Error() != nil {
-		return p
-	}
-	return p.EachLine(func(line string, out *strings.Builder) {
+	return p.FilterLine(func(line string) string {
 		// filepath.Dir() does not handle trailing slashes correctly
 		if len(line) > 1 && strings.HasSuffix(line, "/") {
-			line = line[0 : len(line)-1]
+			line = line[:len(line)-1]
 		}
 		dirname := filepath.Dir(line)
 		// filepath.Dir() does not preserve a leading './'
-		if len(dirname) > 1 && strings.HasPrefix(line, "./") {
-			dirname = "./" + dirname
+		if strings.HasPrefix(line, "./") {
+			return "./" + dirname
 		}
-		out.WriteString(dirname)
-		out.WriteRune('\n')
+		return dirname
 	})
 }
 
 // EachLine calls the specified function for each line of input, passing it the
-// line as a string, and a *strings.Builder to write its output to. The return
-// value from EachLine is a pipe containing the contents of the strings.Builder.
+// line as a string, and a *strings.Builder to write its output to.
+//
+// Deprecated: use FilterLine or FilterScan instead, which run concurrently and
+// don't do unnecessary reads on the input.
 func (p *Pipe) EachLine(process func(string, *strings.Builder)) *Pipe {
-	if p == nil || p.Error() != nil {
-		return p
-	}
-	scanner := bufio.NewScanner(p.Reader)
-	output := strings.Builder{}
-	for scanner.Scan() {
-		process(scanner.Text(), &output)
-		if p.Error() != nil {
-			return p
+	return p.Filter(func(r io.Reader, w io.Writer) error {
+		scanner := bufio.NewScanner(r)
+		output := strings.Builder{}
+		for scanner.Scan() {
+			process(scanner.Text(), &output)
 		}
-	}
-	err := scanner.Err()
-	if err != nil {
-		p.SetError(err)
-	}
-	return Echo(output.String())
+		fmt.Fprint(w, output.String())
+		return scanner.Err()
+	})
 }
 
-// Echo returns a pipe containing the supplied string.
+// Echo produces the supplied string.
 func (p *Pipe) Echo(s string) *Pipe {
-	if p == nil || p.Error() != nil {
+	if p.Error() != nil {
 		return p
 	}
-	return p.WithReader(strings.NewReader(s))
+	return p.WithReader(NewReadAutoCloser(strings.NewReader(s)))
 }
 
 // Exec runs an external command, sending it the contents of the pipe as input,
-// and returns a pipe containing the command's combined output (`stdout` and
-// `stderr`). The effect of this is to use the external command as a filter on
-// the pipe.
+// and produces the command's combined output (`stdout` and `stderr`). The
+// effect of this is to filter the contents of the pipe through the external
+// command.
 //
 // If the command had a non-zero exit status, the pipe's error status will also
 // be set to the string "exit status X", where X is the integer exit status.
-func (p *Pipe) Exec(cmdLine string) *Pipe {
-	if p == nil || p.Error() != nil {
-		return p
-	}
-	q := NewPipe()
-	args, ok := shell.Split(cmdLine) // strings.Fields doesn't handle quotes
-	if !ok {
-		return p.WithError(fmt.Errorf("unbalanced quotes or backslashes in [%s]", cmdLine))
-	}
-	cmd := exec.Command(args[0], args[1:]...)
-	cmd.Stdin = p.Reader
-	output, err := cmd.CombinedOutput()
-	if err != nil {
-		q.SetError(err)
-	}
-	return q.WithReader(bytes.NewReader(output))
+func (p *Pipe) Exec(command string) *Pipe {
+	return p.Filter(func(r io.Reader, w io.Writer) error {
+		args, ok := shell.Split(command) // strings.Fields doesn't handle quotes
+		if !ok {
+			return fmt.Errorf("unbalanced quotes or backslashes in [%s]", command)
+		}
+		cmd := exec.Command(args[0], args[1:]...)
+		cmd.Stdin = r
+		cmd.Stdout = w
+		cmd.Stderr = w
+		err := cmd.Start()
+		if err != nil {
+			fmt.Fprintln(w, err)
+			return err
+		}
+		return cmd.Wait()
+	})
 }
 
 // ExecForEach runs the supplied command once for each line of input, and
-// returns a pipe containing the output. The command string is interpreted as a
-// Go template, so `{{.}}` will be replaced with the input value, for example.
+// produces its combined output. The command string is interpreted as a Go
+// template, so `{{.}}` will be replaced with the input value, for example.
+//
 // If any command resulted in a non-zero exit status, the pipe's error status
 // will also be set to the string "exit status X", where X is the integer exit
 // status.
-func (p *Pipe) ExecForEach(cmdTpl string) *Pipe {
-	if p == nil || p.Error() != nil {
+func (p *Pipe) ExecForEach(command string) *Pipe {
+	if p.Error() != nil {
 		return p
 	}
-	tpl, err := template.New("").Parse(cmdTpl)
+	tpl, err := template.New("").Parse(command)
 	if err != nil {
 		return p.WithError(err)
 	}
-	return p.EachLine(func(line string, out *strings.Builder) {
-		cmdLine := strings.Builder{}
-		err := tpl.Execute(&cmdLine, line)
-		if err != nil {
-			p.SetError(err)
-			return
+	return p.Filter(func(r io.Reader, w io.Writer) error {
+		scanner := bufio.NewScanner(r)
+		for scanner.Scan() {
+			cmdLine := strings.Builder{}
+			err := tpl.Execute(&cmdLine, scanner.Text())
+			if err != nil {
+				return err
+			}
+			args, ok := shell.Split(cmdLine.String()) // strings.Fields doesn't handle quotes
+			if !ok {
+				return fmt.Errorf("unbalanced quotes or backslashes in [%s]", cmdLine.String())
+			}
+			cmd := exec.Command(args[0], args[1:]...)
+			cmd.Stdout = w
+			cmd.Stderr = w
+			err = cmd.Start()
+			if err != nil {
+				fmt.Fprintln(w, err)
+				continue
+			}
+			cmd.Wait()
 		}
-		cmdOutput, err := Exec(cmdLine.String()).String()
-		if err != nil {
-			p.SetError(err)
-			return
+		return scanner.Err()
+	})
+}
+
+// Filter filters the contents of the pipe through the supplied function, which
+// takes an io.Reader (the filter input) and an io.Writer (the filter output),
+// and returns an error, which will be set on the pipe.
+//
+// The filter function runs concurrently, so its goroutine will not complete
+// until the pipe has been fully read. If you just need to make sure all
+// concurrent filters have completed, call Wait on the end of the pipe.
+func (p *Pipe) Filter(filter func(io.Reader, io.Writer) error) *Pipe {
+	pr, pw := io.Pipe()
+	q := NewPipe().WithReader(pr)
+	go func() {
+		defer pw.Close()
+		err := filter(p, pw)
+		q.SetError(err)
+	}()
+	return q
+}
+
+// FilterLine filters the contents of the pipe, a line at a time, through the
+// supplied function, which takes the line as a string and returns a string (the
+// filter output). The filter function runs concurrently.
+func (p *Pipe) FilterLine(filter func(string) string) *Pipe {
+	return p.FilterScan(func(line string, w io.Writer) {
+		fmt.Fprintln(w, filter(line))
+	})
+}
+
+// FilterScan filters the contents of the pipe, a line at a time, through the
+// supplied function, which takes the line as a string and an io.Writer (the
+// filtero output). The filter function runs concurrently.
+func (p *Pipe) FilterScan(filter func(string, io.Writer)) *Pipe {
+	return p.Filter(func(r io.Reader, w io.Writer) error {
+		scanner := bufio.NewScanner(r)
+		for scanner.Scan() {
+			filter(scanner.Text(), w)
 		}
-		out.WriteString(cmdOutput)
+		return scanner.Err()
 	})
 }
 
-// First reads from the pipe, and returns a new pipe containing only the first N
-// lines. If there is an error reading the pipe, the pipe's error status is also
-// set.
-func (p *Pipe) First(lines int) *Pipe {
-	if p == nil || p.Error() != nil {
-		return p
-	}
-	defer p.Close()
-	if lines <= 0 {
+// First produces only the first N lines of input, or the whole input if there
+// are less than N lines. If N is zero or negative, there is no output at all.
+func (p *Pipe) First(n int) *Pipe {
+	if n <= 0 {
 		return NewPipe()
 	}
-	scanner := bufio.NewScanner(p.Reader)
-	output := strings.Builder{}
-	for i := 0; i < lines; i++ {
-		if !scanner.Scan() {
-			break
+	i := 0
+	return p.FilterScan(func(line string, w io.Writer) {
+		if i >= n {
+			return
 		}
-		output.WriteString(scanner.Text())
-		output.WriteRune('\n')
-	}
-	err := scanner.Err()
-	if err != nil {
-		p.SetError(err)
-	}
-	return Echo(output.String())
+		fmt.Fprintln(w, line)
+		i++
+	})
 }
 
-// Freq reads from the pipe, and returns a new pipe containing only unique lines
-// from the input, prefixed with a frequency count, in descending numerical
-// order (most frequent lines first). Lines with equal frequency will be sorted
-// alphabetically. If there is an error reading the pipe, the pipe's error
-// status is also set.
+// Freq produces only unique lines from the input, prefixed with a frequency
+// count, in descending numerical order (most frequent lines first). Lines with
+// equal frequency will be sorted alphabetically.
 //
 // This is a common pattern in shell scripts to find the most
 // frequently-occurring lines in a file:
@@ -547,282 +550,210 @@ func (p *Pipe) First(lines int) *Pipe {
 //  2 orange
 //  1 kumquat
 func (p *Pipe) Freq() *Pipe {
-	if p == nil || p.Error() != nil {
-		return p
-	}
 	freq := map[string]int{}
-	p.EachLine(func(line string, out *strings.Builder) {
-		freq[line]++
-	})
 	type frequency struct {
 		line  string
 		count int
 	}
-	freqs := make([]frequency, 0, len(freq))
-	var maxCount int
-	for line, count := range freq {
-		freqs = append(freqs, frequency{line, count})
-		if count > maxCount {
-			maxCount = count
+	return p.Filter(func(r io.Reader, w io.Writer) error {
+		scanner := bufio.NewScanner(r)
+		for scanner.Scan() {
+			freq[scanner.Text()]++
 		}
-	}
-	sort.Slice(freqs, func(i, j int) bool {
-		if freqs[i].count == freqs[j].count {
-			return freqs[i].line < freqs[j].line
+		freqs := make([]frequency, 0, len(freq))
+		var maxCount int
+		for line, count := range freq {
+			freqs = append(freqs, frequency{line, count})
+			if count > maxCount {
+				maxCount = count
+			}
 		}
-		return freqs[i].count > freqs[j].count
+		sort.Slice(freqs, func(i, j int) bool {
+			x, y := freqs[i].count, freqs[j].count
+			if x == y {
+				return freqs[i].line < freqs[j].line
+			}
+			return x > y
+		})
+		fieldWidth := len(strconv.Itoa(maxCount))
+		for _, item := range freqs {
+			fmt.Fprintf(w, "%*d %s\n", fieldWidth, item.count, item.line)
+		}
+		return nil
 	})
-	fieldWidth := len(strconv.Itoa(maxCount))
-	var output strings.Builder
-	for _, item := range freqs {
-		output.WriteString(fmt.Sprintf("%*d %s", fieldWidth, item.count, item.line))
-		output.WriteRune('\n')
-	}
-	return Echo(output.String())
 }
 
-// Join reads the contents of the pipe, line by line, and joins them into a
-// single space-separated string. It returns a pipe containing this string. Any
-// terminating newline is preserved.
+// Join produces its input as a single space-separated string, which will always
+// end with a newline.
 func (p *Pipe) Join() *Pipe {
-	if p == nil || p.Error() != nil {
-		return p
-	}
-	result, err := p.String()
-	if err != nil {
-		return p
-	}
-	var terminator string
-	if strings.HasSuffix(result, "\n") {
-		terminator = "\n"
-		result = result[:len(result)-1]
-	}
-	output := strings.ReplaceAll(result, "\n", " ")
-	return Echo(output + terminator)
+	return p.Filter(func(r io.Reader, w io.Writer) error {
+		scanner := bufio.NewScanner(r)
+		var line string
+		first := true
+		for scanner.Scan() {
+			if !first {
+				fmt.Fprint(w, " ")
+			}
+			line = scanner.Text()
+			fmt.Fprint(w, line)
+			first = false
+		}
+		fmt.Fprintln(w)
+		return scanner.Err()
+	})
 }
 
-// Last reads from the pipe, and returns a new pipe containing only the last N
-// lines. If there is an error reading the pipe, the pipe's error status is also
-// set.
-func (p *Pipe) Last(lines int) *Pipe {
-	if p == nil || p.Error() != nil {
-		return p
-	}
-	defer p.Close()
-	if lines <= 0 {
+// Last produces only the last N lines of input, or the whole input if there are
+// less than N lines. If N is zero or negative, there is no output at all.
+func (p *Pipe) Last(n int) *Pipe {
+	if n <= 0 {
 		return NewPipe()
 	}
-	scanner := bufio.NewScanner(p.Reader)
-	input := ring.New(lines)
-	for scanner.Scan() {
-		input.Value = scanner.Text()
-		input = input.Next()
-	}
-	output := strings.Builder{}
-	input.Do(func(p interface{}) {
-		value, ok := p.(string)
-		if ok {
-			output.WriteString(value)
-			output.WriteRune('\n')
+	return p.Filter(func(r io.Reader, w io.Writer) error {
+		scanner := bufio.NewScanner(r)
+		input := ring.New(n)
+		for scanner.Scan() {
+			input.Value = scanner.Text()
+			input = input.Next()
 		}
+		input.Do(func(p interface{}) {
+			if p != nil {
+				fmt.Fprintln(w, p)
+			}
+		})
+		return scanner.Err()
 	})
-	err := scanner.Err()
-	if err != nil {
-		p.SetError(err)
-	}
-	return Echo(output.String())
 }
 
-// Match reads from the pipe, and returns a new pipe containing only lines that
-// contain the specified string. If there is an error reading the pipe, the
-// pipe's error status is also set.
+// Match produces only lines that contain the specified string.
 func (p *Pipe) Match(s string) *Pipe {
-	return p.EachLine(func(line string, out *strings.Builder) {
+	return p.FilterScan(func(line string, w io.Writer) {
 		if strings.Contains(line, s) {
-			out.WriteString(line)
-			out.WriteRune('\n')
+			fmt.Fprintln(w, line)
 		}
 	})
 }
 
-// MatchRegexp reads from the pipe, and returns a new pipe containing only lines
-// that match the specified compiled regular expression. If there is an error
-// reading the pipe, the pipe's error status is also set.
+// MatchRegexp produces only lines that match the specified compiled regular
+// expression.
 func (p *Pipe) MatchRegexp(re *regexp.Regexp) *Pipe {
-	if re == nil { // to prevent SIGSEGV
-		return p.WithError(errors.New("nil regular expression"))
-	}
-	return p.EachLine(func(line string, out *strings.Builder) {
+	return p.FilterScan(func(line string, w io.Writer) {
 		if re.MatchString(line) {
-			out.WriteString(line)
-			out.WriteRune('\n')
+			fmt.Fprintln(w, line)
 		}
 	})
 }
 
-// Reject reads from the pipe, and returns a new pipe containing only lines
-// that do not contain the specified string. If there is an error reading the
-// pipe, the pipe's error status is also set.
+// Reject produces only lines that do not contain the specified string.
 func (p *Pipe) Reject(s string) *Pipe {
-	return p.EachLine(func(line string, out *strings.Builder) {
+	return p.FilterScan(func(line string, w io.Writer) {
 		if !strings.Contains(line, s) {
-			out.WriteString(line)
-			out.WriteRune('\n')
+			fmt.Fprintln(w, line)
 		}
 	})
 }
 
-// RejectRegexp reads from the pipe, and returns a new pipe containing only
-// lines that don't match the specified compiled regular expression. If there
-// is an error reading the pipe, the pipe's error status is also set.
+// RejectRegexp produces only lines that don't match the specified compiled
+// regular expression.
 func (p *Pipe) RejectRegexp(re *regexp.Regexp) *Pipe {
-	if re == nil { // to prevent SIGSEGV
-		return p.WithError(errors.New("nil regular expression"))
-	}
-	return p.EachLine(func(line string, out *strings.Builder) {
+	return p.FilterScan(func(line string, w io.Writer) {
 		if !re.MatchString(line) {
-			out.WriteString(line)
-			out.WriteRune('\n')
+			fmt.Fprintln(w, line)
 		}
 	})
 }
 
-// Replace filters its input by replacing all occurrences of the string `search`
-// with the string `replace`. If there is an error reading the pipe, the pipe's
-// error status is also set.
+// Replace replaces all occurrences of the 'search' string with the 'replace'
+// string.
 func (p *Pipe) Replace(search, replace string) *Pipe {
-	return p.EachLine(func(line string, out *strings.Builder) {
-		out.WriteString(strings.ReplaceAll(line, search, replace))
-		out.WriteRune('\n')
+	return p.FilterLine(func(line string) string {
+		return strings.ReplaceAll(line, search, replace)
 	})
 }
 
-// ReplaceRegexp filters its input by replacing all matches of the compiled
-// regular expression `re` with the replacement string `replace`. Inside
-// `replace`, $ signs are interpreted as in regexp.Expand, so for instance "$1"
-// represents the text of the first submatch. If there is an error reading the
-// pipe, the pipe's error status is also set.
+// ReplaceRegexp replaces all matches of the specified compiled regular
+// expression with the 'replace' string. '$' characters in the replace string
+// are interpreted as in regexp.Expand; for example, "$1" represents the text of
+// the first submatch.
 func (p *Pipe) ReplaceRegexp(re *regexp.Regexp, replace string) *Pipe {
-	if re == nil { // to prevent SIGSEGV
-		return p.WithError(errors.New("nil regular expression"))
-	}
-	return p.EachLine(func(line string, out *strings.Builder) {
-		out.WriteString(re.ReplaceAllString(line, replace))
-		out.WriteRune('\n')
+	return p.FilterLine(func(line string) string {
+		return re.ReplaceAllString(line, replace)
 	})
 }
 
 // SHA256Sums reads a list of file paths from the pipe, one per line, and
-// returns a pipe that contains the SHA-256 checksum of each pathname, in hex.
-// If there are any errors (for example, non-existent files), the pipe's error
-// status will be set to the first error encountered, but execution will
-// continue.
+// produces the hex-encoded SHA-256 hash of each file. Any files that cannot be
+// opened or read will be ignored.
 func (p *Pipe) SHA256Sums() *Pipe {
-	if p == nil || p.Error() != nil {
-		return p
-	}
-
-	return p.EachLine(func(line string, out *strings.Builder) {
+	return p.FilterScan(func(line string, w io.Writer) {
 		f, err := os.Open(line)
 		if err != nil {
-			p.SetError(err)
-			return
+			return // skip unopenable files
 		}
 		defer f.Close()
-
 		h := sha256.New()
-		if _, err := io.Copy(h, f); err != nil {
-			p.SetError(err)
-			return
+		_, err = io.Copy(h, f)
+		if err != nil {
+			return // skip unreadable files
 		}
-
-		out.WriteString(hex.EncodeToString(h.Sum(nil)[:]))
-		out.WriteRune('\n')
+		fmt.Fprintln(w, hex.EncodeToString(h.Sum(nil)))
 	})
 }
 
-// AppendFile appends the contents of the Pipe to the specified file, and closes
-// the pipe after reading. If the file does not exist, it is created.
-//
-// AppendFile returns the number of bytes successfully written, or an error. If
-// there is an error reading or writing, the pipe's error status is also set.
+// AppendFile appends the contents of the pipe to the specified file, and
+// returns the number of bytes successfully written, or an error. If the file
+// does not exist, it is created.
 func (p *Pipe) AppendFile(fileName string) (int64, error) {
 	return p.writeOrAppendFile(fileName, os.O_APPEND|os.O_CREATE|os.O_WRONLY)
 }
 
-// Bytes returns the contents of the Pipe as a slice of byte, or an error. If
-// there is an error reading, the pipe's error status is also set.
+// Bytes returns the contents of the pipe as a []]byte, or an error.
 func (p *Pipe) Bytes() ([]byte, error) {
-	if p == nil || p.Error() != nil {
-		return []byte{}, p.Error()
-	}
-	res, err := ioutil.ReadAll(p.Reader)
+	res, err := io.ReadAll(p)
 	if err != nil {
 		p.SetError(err)
-		return []byte{}, err
 	}
-	return res, nil
+	return res, err
 }
 
-// CountLines counts lines from the pipe's reader, and returns the integer
-// result, or an error. If there is an error reading the pipe, the pipe's error
-// status is also set.
+// CountLines returns the number of lines of input, or an error.
 func (p *Pipe) CountLines() (int, error) {
-	if p == nil || p.Error() != nil {
-		return 0, p.Error()
-	}
-	var lines int
-	p.EachLine(func(line string, out *strings.Builder) {
+	lines := 0
+	p.FilterScan(func(line string, w io.Writer) {
 		lines++
-	})
+	}).Wait()
 	return lines, p.Error()
 }
 
-// SHA256Sum calculates the SHA-256 of the file from the pipe's reader, and
-// returns the hex-encoded string result, or an error. If there is an error
-// reading the pipe, the pipe's error status is also set.
+// SHA256Sum returns the hex-encoded SHA-256 hash of its input, or an error.
 func (p *Pipe) SHA256Sum() (string, error) {
-	if p == nil || p.Error() != nil {
-		return "", p.Error()
-	}
-
-	h := sha256.New()
-	if _, err := io.Copy(h, p.Reader); err != nil {
+	hasher := sha256.New()
+	_, err := io.Copy(hasher, p)
+	if err != nil {
 		p.SetError(err)
-		return "", p.Error()
+		return "", err
 	}
-
-	encodedCheckSum := hex.EncodeToString(h.Sum(nil)[:])
-	return encodedCheckSum, nil
+	return hex.EncodeToString(hasher.Sum(nil)), p.Error()
 }
 
-// Slice returns the contents of the pipe as a slice of strings, one element per
-// line, or an error. If there is an error reading the pipe, the pipe's error
-// status is also set.
+// Slice returns the input as a slice of strings, one element per line, or an
+// error.
 //
 // An empty pipe will produce an empty slice. A pipe containing a single empty
-// line (that is, a single `\n` character) will produce a slice of one element
-// that is the empty string.
+// line (that is, a single `\n` character) will produce a slice containing the
+// empty string.
 func (p *Pipe) Slice() ([]string, error) {
-	if p == nil || p.Error() != nil {
-		return nil, p.Error()
-	}
 	result := []string{}
-	p.EachLine(func(line string, out *strings.Builder) {
+	p.FilterScan(func(line string, w io.Writer) {
 		result = append(result, line)
-	})
+	}).Wait()
 	return result, p.Error()
 }
 
-// Stdout writes the contents of the pipe to its configured standard output. It
-// returns the number of bytes successfully written, plus a non-nil error if the
-// write failed or if there was an error reading from the pipe. If the pipe has
-// error status, Stdout returns zero plus the existing error.
+// Stdout writes the input to the pipe's configured standard output, and returns
+// the number of bytes successfully written, or an error.
 func (p *Pipe) Stdout() (int, error) {
-	if p == nil || p.Error() != nil || p.stdout == nil {
-		return 0, p.Error()
-	}
-	n64, err := io.Copy(p.stdout, p.Reader)
+	n64, err := io.Copy(p.stdout, p)
 	if err != nil {
 		return 0, err
 	}
@@ -830,45 +761,42 @@ func (p *Pipe) Stdout() (int, error) {
 	if int64(n) != n64 {
 		return 0, fmt.Errorf("length %d overflows int", n64)
 	}
-	return n, nil
+	return n, p.Error()
 }
 
-// String returns the contents of the Pipe as a string, or an error, and closes
-// the pipe after reading. If there is an error reading, the pipe's error status
-// is also set.
-//
-// Note that String consumes the complete output of the pipe, which closes the
-// input reader automatically. Therefore, calling String (or any other sink
-// method) again on the same pipe will return an error.
+// String returns the input as a string, or an error.
 func (p *Pipe) String() (string, error) {
 	data, err := p.Bytes()
 	if err != nil {
 		p.SetError(err)
-		return "", err
 	}
-	return string(data), nil
+	return string(data), p.Error()
 }
 
-// WriteFile writes the contents of the Pipe to the specified file, and closes
-// the pipe after reading. If the file already exists, it is truncated and the
-// new data will replace the old. It returns the number of bytes successfully
-// written, or an error. If there is an error reading or writing, the pipe's
-// error status is also set.
+// Wait reads the input to completion and discards it. This is mostly useful for
+// waiting until all concurrent filter stages have finished.
+func (p *Pipe) Wait() {
+	_, err := io.Copy(io.Discard, p)
+	if err != nil {
+		p.SetError(err)
+	}
+}
+
+// WriteFile writes the input to the specified file, and returns the number of
+// bytes successfully written, or an error. If the file already exists, it is
+// truncated and the new data will replace the old.
 func (p *Pipe) WriteFile(fileName string) (int64, error) {
 	return p.writeOrAppendFile(fileName, os.O_RDWR|os.O_CREATE|os.O_TRUNC)
 }
 
 func (p *Pipe) writeOrAppendFile(fileName string, mode int) (int64, error) {
-	if p == nil || p.Error() != nil {
-		return 0, p.Error()
-	}
 	out, err := os.OpenFile(fileName, mode, 0666)
 	if err != nil {
 		p.SetError(err)
 		return 0, err
 	}
 	defer out.Close()
-	wrote, err := io.Copy(out, p.Reader)
+	wrote, err := io.Copy(out, p)
 	if err != nil {
 		p.SetError(err)
 		return 0, err


Test patch:
diff --git a/script_test.go b/script_test.go
index 52234a5..4e5c60b 100644
--- a/script_test.go
+++ b/script_test.go
@@ -1,15 +1,16 @@
 package script_test
 
 import (
+	"bufio"
 	"bytes"
 	"errors"
 	"fmt"
 	"io"
-	"io/ioutil"
 	"os"
 	"os/exec"
 	"path/filepath"
 	"regexp"
+	"runtime"
 	"strings"
 	"testing"
 
@@ -30,7 +31,7 @@ func TestMain(m *testing.M) {
 	}
 }
 
-func TestArgs(t *testing.T) {
+func TestArgsSuppliesCommandLineArgumentsAsInputToPipeOnePerLine(t *testing.T) {
 	t.Parallel()
 	// dummy test to prove coverage
 	script.Args()
@@ -47,20 +48,7 @@ func TestArgs(t *testing.T) {
 	}
 }
 
-func TestEcho(t *testing.T) {
-	t.Parallel()
-	want := "Hello, world."
-	p := script.Echo(want)
-	got, err := p.String()
-	if err != nil {
-		t.Error(err)
-	}
-	if got != want {
-		t.Errorf("want %q, got %q", want, got)
-	}
-}
-
-func TestBasename(t *testing.T) {
+func TestBasenameRemovesLeadingPathComponentsFromInputLines(t *testing.T) {
 	t.Parallel()
 	tcs := []struct {
 		path string
@@ -224,13 +212,14 @@ func TestConcatOutputsContentsOfSpecifiedFilesInOrder(t *testing.T) {
 	}
 }
 
-func TestDirname(t *testing.T) {
+func TestDirname_RemovesFilenameComponentFromInputLines(t *testing.T) {
 	t.Parallel()
 	tcs := []struct {
 		path string
 		want string
 	}{
 		{"/", "/\n"},
+		{"./", ".\n"},
 		{"\n", ".\n"},
 		{"/root", "/\n"},
 		{"/tmp/example.php", "/tmp\n"},
@@ -256,7 +245,7 @@ func TestDirname(t *testing.T) {
 	}
 }
 
-func TestEachLine(t *testing.T) {
+func TestEachLine_FiltersInputThroughSuppliedFunction(t *testing.T) {
 	t.Parallel()
 	p := script.Echo("Hello\nGoodbye")
 	q := p.EachLine(func(line string, out *strings.Builder) {
@@ -272,10 +261,23 @@ func TestEachLine(t *testing.T) {
 	}
 }
 
-func TestEchoFilter(t *testing.T) {
+func TestEchoProducesSuppliedString(t *testing.T) {
 	t.Parallel()
 	want := "Hello, world."
-	p := script.NewPipe().Echo(want)
+	p := script.Echo(want)
+	got, err := p.String()
+	if err != nil {
+		t.Error(err)
+	}
+	if got != want {
+		t.Errorf("want %q, got %q", want, got)
+	}
+}
+
+func TestEchoReplacesInputWithSuppliedStringWhenUsedAsFilter(t *testing.T) {
+	t.Parallel()
+	want := "Hello, world."
+	p := script.Echo("bogus").Echo(want)
 	got, err := p.String()
 	if err != nil {
 		t.Error(err)
@@ -287,13 +289,163 @@ func TestEchoFilter(t *testing.T) {
 
 func TestExecForEach_ErrorsOnInvalidTemplateSyntax(t *testing.T) {
 	t.Parallel()
-	p := script.Echo("a\nb\nc\n")
-	p.ExecForEach("{{invalid template syntax}}")
+	p := script.Echo("a\nb\nc\n").ExecForEach("{{invalid template syntax}}")
+	p.Wait()
 	if p.Error() == nil {
 		t.Error("want error with invalid template syntax")
 	}
 }
 
+func TestExecForEach_ErrorsOnUnbalancedQuotes(t *testing.T) {
+	t.Parallel()
+	p := script.Echo("a\nb\nc\n").ExecForEach("echo \"{{.}}")
+	p.Wait()
+	if p.Error() == nil {
+		t.Error("want error with unbalanced quotes in command line")
+	}
+}
+
+func TestFilterByCopyPassesInputThroughUnchanged(t *testing.T) {
+	t.Parallel()
+	p := script.Echo("hello").Filter(func(r io.Reader, w io.Writer) error {
+		_, err := io.Copy(w, r)
+		return err
+	})
+	want := "hello"
+	got, err := p.String()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if want != got {
+		t.Error(cmp.Diff(want, got))
+	}
+}
+
+func TestFilterCanChainFilters(t *testing.T) {
+	t.Parallel()
+	p := script.Echo("hello").Filter(func(r io.Reader, w io.Writer) error {
+		_, err := io.Copy(w, r)
+		return err
+	}).Filter(func(r io.Reader, w io.Writer) error {
+		_, err := io.Copy(w, r)
+		return err
+	})
+	want := "hello"
+	got, err := p.String()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if want != got {
+		t.Error(cmp.Diff(want, got))
+	}
+}
+
+func TestFilterByCopyToDiscardGivesNoOutput(t *testing.T) {
+	t.Parallel()
+	p := script.Echo("hello").Filter(func(r io.Reader, w io.Writer) error {
+		_, err := io.Copy(io.Discard, r)
+		return err
+	})
+	want := ""
+	got, err := p.String()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if want != got {
+		t.Error(cmp.Diff(want, got))
+	}
+}
+
+func TestFilterReadsNoMoreThanRequested(t *testing.T) {
+	t.Parallel()
+	input := "firstline\nsecondline"
+	source := bytes.NewBufferString(input)
+	p := script.NewPipe().WithReader(source).Filter(func(r io.Reader, w io.Writer) error {
+		// read just the first line of input
+		var text string
+		_, err := fmt.Fscanln(r, &text)
+		if err != nil {
+			return err
+		}
+		fmt.Fprintln(w, text)
+		return nil
+	})
+	runtime.Gosched() // give filter goroutine a chance to run
+	want := "firstline\n"
+	got, err := p.String()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if want != got {
+		t.Error(cmp.Diff(want, got))
+	}
+	wantRemaining := "secondline"
+	if wantRemaining != source.String() {
+		t.Errorf("want %q remaining, got %q", wantRemaining, source.String())
+	}
+}
+
+func TestFilterByFirstLineOnlyGivesFirstLineOfInput(t *testing.T) {
+	t.Parallel()
+	p := script.Echo("hello\nworld").Filter(func(r io.Reader, w io.Writer) error {
+		scanner := bufio.NewScanner(r)
+		for scanner.Scan() {
+			fmt.Fprintln(w, scanner.Text())
+			break
+		}
+		return scanner.Err()
+	})
+	want := "hello\n"
+	got, err := p.String()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if want != got {
+		t.Error(cmp.Diff(want, got))
+	}
+}
+
+func TestFilterSetsErrorOnPipeIfFilterFuncReturnsError(t *testing.T) {
+	t.Parallel()
+	p := script.Echo("hello").Filter(func(io.Reader, io.Writer) error {
+		return errors.New("oh no")
+	})
+	io.ReadAll(p)
+	if p.Error() == nil {
+		t.Error("no error")
+	}
+}
+
+func TestFilterLine_FiltersEachLineThroughSuppliedFunction(t *testing.T) {
+	t.Parallel()
+	input := "hello\nworld"
+	want := "HELLO\nWORLD\n"
+	got, err := script.Echo(input).FilterLine(strings.ToUpper).String()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if want != got {
+		t.Error(cmp.Diff(want, got))
+	}
+}
+
+func TestFilterScan_FiltersInputLineByLine(t *testing.T) {
+	t.Parallel()
+	input := "hello\nworld\ngoodbye"
+	want := "world\n"
+	got, err := script.Echo(input).FilterScan(func(line string, w io.Writer) {
+		if strings.HasPrefix(line, "w") {
+			fmt.Fprintln(w, line)
+		}
+	}).String()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if want != got {
+		t.Error(cmp.Diff(want, got))
+	}
+}
+
 func TestFirstDropsAllButFirstNLinesOfInput(t *testing.T) {
 	t.Parallel()
 	input := "a\nb\nc\n"
@@ -379,9 +531,9 @@ func TestFreqProducesCorrectFrequencyTableForInput(t *testing.T) {
 	}
 }
 
-func TestJoin(t *testing.T) {
+func TestJoinJoinsInputLinesIntoSpaceSeparatedString(t *testing.T) {
 	t.Parallel()
-	input := "hello\nfrom\nthe\njoin\ntest\n"
+	input := "hello\nfrom\nthe\njoin\ntest"
 	want := "hello from the join test\n"
 	got, err := script.Echo(input).Join().String()
 	if err != nil {
@@ -390,15 +542,6 @@ func TestJoin(t *testing.T) {
 	if got != want {
 		t.Errorf("want %q, got %q", want, got)
 	}
-	input = "hello\nworld"
-	want = "hello world"
-	got, err = script.Echo(input).Join().String()
-	if err != nil {
-		t.Error(err)
-	}
-	if got != want {
-		t.Errorf("want %q, got %q", want, got)
-	}
 }
 
 func TestLastDropsAllButLastNLinesOfInput(t *testing.T) {
@@ -494,14 +637,6 @@ func TestMatchOutputsNothingGivenEmptyInput(t *testing.T) {
 	}
 }
 
-func TestMatchRegexp_ErrorsGivenANilRegexp(t *testing.T) {
-	t.Parallel()
-	p := script.NewPipe().MatchRegexp(nil)
-	if p.Error() == nil {
-		t.Error("want error on nil regexp")
-	}
-}
-
 func TestMatchRegexp_OutputsOnlyLinesMatchingRegexp(t *testing.T) {
 	t.Parallel()
 	input := "This is the first line in the file.\nHello, world.\nThis is another line in the file.\n"
@@ -705,6 +840,7 @@ func TestSHA256Sums_OutputsCorrectHashForEachSpecifiedFile(t *testing.T) {
 func TestExecErrorsWhenTheSpecifiedCommandDoesNotExist(t *testing.T) {
 	t.Parallel()
 	p := script.Exec("doesntexist")
+	p.Wait()
 	if p.Error() == nil {
 		t.Error("want error running non-existent command")
 	}
@@ -715,17 +851,12 @@ func TestExecRunsGoWithNoArgsAndGetsUsageMessagePlusErrorExitStatus2(t *testing.
 	// We can't make many cross-platform assumptions about what external
 	// commands will be available, but it seems logical that 'go' would be
 	// (though it may not be in the user's path)
-	p := script.Exec("go")
-	if p.Error() == nil {
-		t.Error("want error running 'go' with no arguments (because this command returns a non-zero exit status")
-	}
-	p.SetError(nil) // else p.String() would be a no-op
-	output, err := p.String()
-	if err != nil {
-		t.Fatal(err)
+	got, err := script.Exec("go").String()
+	if err == nil {
+		t.Error("want error when command returns a non-zero exit status")
 	}
-	if !strings.Contains(output, "Usage") {
-		t.Fatalf("want output containing the word 'usage', got %q", output)
+	if !strings.Contains(got, "Usage") {
+		t.Fatalf("want output containing the word 'usage', got %q", got)
 	}
 }
 
@@ -808,15 +939,27 @@ func TestFindFiles_InNonexistentPathReturnsError(t *testing.T) {
 	}
 }
 
-func TestIfExists(t *testing.T) {
+func TestIfExists_ProducesErrorPlusNoOutputForNonexistentFile(t *testing.T) {
 	t.Parallel()
-	p := script.IfExists("testdata/doesntexist")
-	if p.Error() == nil {
-		t.Errorf("want error from IfExists on non-existent file, but got nil")
+	want := ""
+	got, err := script.IfExists("testdata/doesntexist").Echo("hello").String()
+	if err == nil {
+		t.Error("want error")
 	}
-	p = script.IfExists("testdata/empty.txt")
-	if p.Error() != nil {
-		t.Errorf("want no error from IfExists on existing file, but got %v", p.Error())
+	if want != got {
+		t.Error(cmp.Diff(want, got))
+	}
+}
+
+func TestIfExists_ProducesOutputAndNoErrorWhenFileExists(t *testing.T) {
+	t.Parallel()
+	want := "hello"
+	got, err := script.IfExists("testdata/empty.txt").Echo("hello").String()
+	if err != nil {
+		t.Error(err)
+	}
+	if want != got {
+		t.Error(cmp.Diff(want, got))
 	}
 }
 
@@ -885,7 +1028,7 @@ func TestReadAutoCloser_ReadsAllDataFromSourceAndClosesItAutomatically(t *testin
 	}
 }
 
-func TestSlice(t *testing.T) {
+func TestSliceProducesElementsOfSpecifiedSliceOnePerLine(t *testing.T) {
 	t.Parallel()
 	want := "1\n2\n3\n"
 	got, err := script.Slice([]string{"1", "2", "3"}).String()
@@ -897,7 +1040,7 @@ func TestSlice(t *testing.T) {
 	}
 }
 
-func TestStdin(t *testing.T) {
+func TestStdinReadsFromProgramStandardInput(t *testing.T) {
 	t.Parallel()
 	// dummy test to prove coverage
 	script.Stdin()
@@ -905,7 +1048,7 @@ func TestStdin(t *testing.T) {
 	want := "hello world"
 	cmd := exec.Command(os.Args[0])
 	cmd.Env = append(os.Environ(), "SCRIPT_TEST=stdin")
-	cmd.Stdin = script.Echo(want).Reader
+	cmd.Stdin = script.Echo(want)
 	got, err := cmd.Output()
 	if err != nil {
 		t.Error(err)
@@ -915,63 +1058,36 @@ func TestStdin(t *testing.T) {
 	}
 }
 
-func TestSinksOnNilPipes(t *testing.T) {
-	t.Parallel()
-	doSinksOnPipe(t, nil, "nil")
-}
-
-func TestSinksOnZeroPipes(t *testing.T) {
+func TestStdoutSendsPipeContentsToConfiguredStandardOutput(t *testing.T) {
 	t.Parallel()
-	doSinksOnPipe(t, &script.Pipe{}, "zero")
-}
-
-// doSinksOnPipe calls every kind of sink method on the supplied pipe and
-// tries to trigger a panic.
-func doSinksOnPipe(t *testing.T, p *script.Pipe, kind string) {
-	var action string
-	defer func() {
-		if r := recover(); r != nil {
-			t.Errorf("panic: %s on %s pipe", action, kind)
-		}
-	}()
-	action = "String()"
-	_, err := p.String()
-	if err != nil {
-		t.Error(err)
-	}
-	action = "CountLines()"
-	_, err = p.CountLines()
-	if err != nil {
-		t.Error(err)
-	}
-	action = "SHA256Sum()"
-	_, err = p.SHA256Sum()
+	buf := &bytes.Buffer{}
+	want := "hello world"
+	p := script.File("testdata/hello.txt").WithStdout(buf)
+	wrote, err := p.Stdout()
 	if err != nil {
 		t.Error(err)
 	}
-	action = "Slice()"
-	_, err = p.Slice()
-	if err != nil {
-		t.Error(err)
+	if wrote != len(want) {
+		t.Errorf("want %d bytes written, got %d", len(want), wrote)
 	}
-	action = "WriteFile()"
-	_, err = p.WriteFile(t.TempDir() + "/" + kind)
-	if err != nil {
-		t.Error(err)
+	got := buf.String()
+	if want != got {
+		t.Errorf("want %q, got %q", want, string(got))
 	}
-	action = "AppendFile()"
-	_, err = p.AppendFile(t.TempDir() + "/" + kind)
-	if err != nil {
-		t.Error(err)
+	_, err = p.String()
+	if err == nil {
+		t.Error("input not closed after reading")
 	}
 }
 
-func TestAppendFile(t *testing.T) {
+func TestAppendFile_AppendsAllItsInputToSpecifiedFile(t *testing.T) {
 	t.Parallel()
 	orig := "Hello, world"
 	path := t.TempDir() + "/" + t.Name()
-	// ignore results; we're testing AppendFile, not WriteFile
-	_, _ = script.Echo(orig).WriteFile(path)
+	_, err := script.Echo(orig).WriteFile(path)
+	if err != nil {
+		t.Fatal(err)
+	}
 	extra := " and goodbye"
 	wrote, err := script.Echo(extra).AppendFile(path)
 	if err != nil {
@@ -1003,43 +1119,27 @@ func TestBytesOutputsInputBytesUnchanged(t *testing.T) {
 	}
 }
 
-func TestCountLines(t *testing.T) {
+func TestCountLines_CountsCorrectNumberOfLinesInInput(t *testing.T) {
 	t.Parallel()
 	want := 3
-	got, err := script.File("testdata/test.txt").CountLines()
-	if err != nil {
-		t.Error(err)
-	}
-	if got != want {
-		t.Errorf("counting non-empty file: want %d, got %d", want, got)
-	}
-	want = 0
-	got, err = script.File("testdata/empty.txt").CountLines()
+	got, err := script.Echo("a\nb\nc").CountLines()
 	if err != nil {
-		t.Error(err)
+		t.Fatal(err)
 	}
 	if got != want {
-		t.Errorf("counting empty file: want %d, got %d", want, got)
+		t.Errorf("want %d, got %d", want, got)
 	}
-	want = 3
-	p := script.File("testdata/test.txt")
-	got, err = p.CountLines()
+}
+
+func TestCountLines_Counts0LinesInEmptyInput(t *testing.T) {
+	t.Parallel()
+	want := 0
+	got, err := script.Echo("").CountLines()
 	if err != nil {
-		t.Error(err)
+		t.Fatal(err)
 	}
 	if got != want {
-		t.Errorf("want %d lines, got %d", want, got)
-	}
-	_, err = ioutil.ReadAll(p.Reader)
-	if err == nil {
-		t.Error("input not closed after reading")
-	}
-	_, err = p.CountLines() // result should be zero
-	if p.Error() == nil {
-		t.Error("want error reading closed pipe, got nil")
-	}
-	if err != p.Error() {
-		t.Errorf("got error %v but pipe error status was %v", err, p.Error())
+		t.Errorf("want %d, got %d", want, got)
 	}
 }
 
@@ -1078,54 +1178,48 @@ func TestSHA256Sum_OutputsCorrectHash(t *testing.T) {
 	}
 }
 
-func TestSliceSink(t *testing.T) {
+func TestSliceSink_(t *testing.T) {
 	t.Parallel()
 	tests := []struct {
-		name    string
-		fields  *script.Pipe
-		want    []string
-		wantErr bool
+		name string
+		pipe *script.Pipe
+		want []string
 	}{
 		{
-			name:   "Multiple lines pipe",
-			fields: script.Echo("testdata/multiple_files/1.txt\ntestdata/multiple_files/2.txt\ntestdata/multiple_files/3.tar.zip\n"),
+			name: "returns three elements for three lines of input",
+			pipe: script.Echo("testdata/multiple_files/1.txt\ntestdata/multiple_files/2.txt\ntestdata/multiple_files/3.tar.zip\n"),
 			want: []string{
 				"testdata/multiple_files/1.txt",
 				"testdata/multiple_files/2.txt",
 				"testdata/multiple_files/3.tar.zip",
 			},
-			wantErr: false,
 		},
 		{
-			name:    "Empty pipe",
-			fields:  script.Echo(""),
-			want:    []string{},
-			wantErr: false,
+			name: "returns an empty slice given empty input",
+			pipe: script.Echo(""),
+			want: []string{},
 		},
 		{
-			name:    "Single newline",
-			fields:  script.Echo("\n"),
-			want:    []string{""},
-			wantErr: false,
+			name: "returns one empty string given input containing a single newline",
+			pipe: script.Echo("\n"),
+			want: []string{""},
 		},
 		{
-			name:   "Empty line between two existing lines",
-			fields: script.Echo("testdata/multiple_files/1.txt\n\ntestdata/multiple_files/3.tar.zip"),
+			name: "returns an empty string for each empty input line",
+			pipe: script.Echo("testdata/multiple_files/1.txt\n\ntestdata/multiple_files/3.tar.zip"),
 			want: []string{
 				"testdata/multiple_files/1.txt",
 				"",
 				"testdata/multiple_files/3.tar.zip",
 			},
-			wantErr: false,
 		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			p := tt.fields
+			p := tt.pipe
 			got, err := p.Slice()
-			if (err != nil) != tt.wantErr {
-				t.Errorf("Slice() error = %v, wantErr %v", err, tt.wantErr)
-				return
+			if err != nil {
+				t.Fatal(err)
 			}
 			if !cmp.Equal(tt.want, got) {
 				t.Error(cmp.Diff(tt.want, got))
@@ -1134,46 +1228,6 @@ func TestSliceSink(t *testing.T) {
 	}
 }
 
-func TestStdout(t *testing.T) {
-	t.Parallel()
-	buf := &bytes.Buffer{}
-	want := "hello world"
-	p := script.File("testdata/hello.txt").WithStdout(buf)
-	wrote, err := p.Stdout()
-	if err != nil {
-		t.Error(err)
-	}
-	if wrote != len(want) {
-		t.Errorf("want %d bytes written, got %d", len(want), wrote)
-	}
-	got := buf.String()
-	if want != got {
-		t.Errorf("want %q, got %q", want, string(got))
-	}
-	_, err = p.String()
-	if err == nil {
-		t.Error("input not closed after reading")
-	}
-}
-
-func TestStdoutNoPanicOnNilOrZero(t *testing.T) {
-	t.Parallel()
-	kind := "nil pipe"
-	defer func() {
-		if r := recover(); r != nil {
-			t.Fatalf("panic: Stdout on %s", kind)
-		}
-	}()
-	var p *script.Pipe
-	_, _ = p.Stdout()
-	kind = "zero pipe"
-	p = &script.Pipe{}
-	_, _ = p.Stdout()
-	kind = "zero pipe with non-empty reader"
-	p.Reader = script.NewReadAutoCloser(strings.NewReader("bogus"))
-	_, _ = p.Stdout()
-}
-
 func TestStringOutputsInputStringUnchanged(t *testing.T) {
 	t.Parallel()
 	want := "hello, world"
@@ -1186,7 +1240,16 @@ func TestStringOutputsInputStringUnchanged(t *testing.T) {
 	}
 }
 
-func TestWriteFileNew(t *testing.T) {
+func TestWaitReadsPipeSourceToCompletion(t *testing.T) {
+	t.Parallel()
+	source := bytes.NewBufferString("hello")
+	script.NewPipe().WithReader(source).FilterLine(strings.ToUpper).Wait()
+	if source.Len() > 0 {
+		t.Errorf("incomplete read: %d bytes of input remaining: %q", source.Len(), source.String())
+	}
+}
+
+func TestWriteFile_WritesInputToFileCreatingItIfNecessary(t *testing.T) {
 	t.Parallel()
 	want := "Hello, world"
 	path := t.TempDir() + "/" + t.Name()
@@ -1206,7 +1269,7 @@ func TestWriteFileNew(t *testing.T) {
 	}
 }
 
-func TestWriteFileTruncatesExisting(t *testing.T) {
+func TestWriteFile_TruncatesExistingFile(t *testing.T) {
 	t.Parallel()
 	want := "Hello, world"
 	path := t.TempDir() + "/" + t.Name()
@@ -1235,7 +1298,7 @@ func TestWriteFileTruncatesExisting(t *testing.T) {
 	}
 }
 
-func TestWithReader(t *testing.T) {
+func TestWithReader_SetsSuppliedReaderOnPipe(t *testing.T) {
 	t.Parallel()
 	want := "Hello, world."
 	p := script.NewPipe().WithReader(strings.NewReader(want))
@@ -1248,26 +1311,16 @@ func TestWithReader(t *testing.T) {
 	}
 }
 
-func TestWithError(t *testing.T) {
+func TestWithError_SetsSpecifiedErrorOnPipe(t *testing.T) {
 	t.Parallel()
-	p := script.File("testdata/empty.txt")
-	want := "fake error"
-	_, gotErr := p.WithError(errors.New(want)).String()
-	if gotErr.Error() != "fake error" {
-		t.Errorf("want %q, got %q", want, gotErr.Error())
-	}
-	_, err := ioutil.ReadAll(p.Reader)
-	if err == nil {
-		t.Error("input not closed after reading")
-	}
-	p = script.File("testdata/empty.txt")
-	_, gotErr = p.WithError(nil).String()
-	if gotErr != nil {
-		t.Errorf("got unexpected error: %q", gotErr.Error())
+	fakeErr := errors.New("oh no")
+	p := script.NewPipe().WithError(fakeErr)
+	if p.Error() != fakeErr {
+		t.Errorf("want %q, got %q", fakeErr, p.Error())
 	}
 }
 
-func TestWithStdout(t *testing.T) {
+func TestWithStdout_SetsSpecifiedWriterAsStdout(t *testing.T) {
 	t.Parallel()
 	buf := &bytes.Buffer{}
 	want := "Hello, world."
@@ -1281,26 +1334,17 @@ func TestWithStdout(t *testing.T) {
 	}
 }
 
-func TestError(t *testing.T) {
+func TestErrorReturnsErrorSetByPreviousPipeStage(t *testing.T) {
 	t.Parallel()
 	p := script.File("testdata/nonexistent.txt")
 	if p.Error() == nil {
 		t.Error("want error status reading nonexistent file, but got nil")
 	}
-	defer func() {
-		// Reading an erroneous pipe should not panic.
-		if r := recover(); r != nil {
-			t.Errorf("panic reading erroneous pipe: %v", r)
-		}
-	}()
-	_, err := p.String()
-	if err != p.Error() {
-		t.Error(err)
-	}
-	_, err = p.CountLines()
-	if err != p.Error() {
-		t.Error(err)
-	}
+}
+
+func TestSetError_SetsSuppliedErrorOnPipe(t *testing.T) {
+	t.Parallel()
+	p := script.NewPipe()
 	e := errors.New("fake error")
 	p.SetError(e)
 	if p.Error() != e {
@@ -1308,7 +1352,7 @@ func TestError(t *testing.T) {
 	}
 }
 
-func TestExitStatus(t *testing.T) {
+func TestExitStatus_CorrectlyParsesExitStatusValueFromErrorMessage(t *testing.T) {
 	t.Parallel()
 	tcs := []struct {
 		input string
@@ -1336,106 +1380,29 @@ func TestExitStatus(t *testing.T) {
 	}
 }
 
-// doMethodsOnPipe calls every kind of method on the supplied pipe and
-// tries to trigger a panic.
-func doMethodsOnPipe(t *testing.T, p *script.Pipe, kind string) {
-	var action string
-	defer func() {
-		if r := recover(); r != nil {
-			t.Errorf("panic: %s on %s pipe", action, kind)
-		}
-	}()
-	action = "AppendFile()"
-	p.AppendFile(t.TempDir() + "/AppendFile")
-	action = "Basename()"
-	p.Basename()
-	action = "Bytes()"
-	p.Bytes()
-	action = "Close()"
-	p.Close()
-	action = "Column()"
-	p.Column(2)
-	action = "Concat()"
-	p.Concat()
-	action = "CountLines()"
-	p.CountLines()
-	action = "Dirname()"
-	p.Dirname()
-	action = "EachLine()"
-	p.EachLine(func(string, *strings.Builder) {})
-	action = "Echo()"
-	p.Echo("bogus")
-	action = "Error()"
-	p.Error()
-	action = "Exec()"
-	p.Exec("bogus")
-	action = "ExecForEach()"
-	p.ExecForEach("bogus")
-	action = "ExitStatus()"
-	p.ExitStatus()
-	action = "First()"
-	p.First(1)
-	action = "Freq()"
-	p.Freq()
-	action = "Join()"
-	p.Join()
-	action = "Last()"
-	p.Last(1)
-	action = "Match()"
-	p.Match("foo")
-	action = "MatchRegexp()"
-	p.MatchRegexp(regexp.MustCompile(".*"))
-	action = "Read()"
-	p.Read([]byte{})
-	action = "Reject()"
-	p.Reject("")
-	action = "RejectRegexp"
-	p.RejectRegexp(regexp.MustCompile(".*"))
-	action = "Replace()"
-	p.Replace("old", "new")
-	action = "ReplaceRegexp()"
-	p.ReplaceRegexp(regexp.MustCompile(".*"), "")
-	action = "SetError()"
-	p.SetError(nil)
-	action = "SHA256Sums()"
-	p.SHA256Sums()
-	action = "SHA256Sum()"
-	p.SHA256Sum()
-	action = "Slice()"
-	p.Slice()
-	action = "Stdout()"
-	p.Stdout()
-	action = "String()"
-	p.String()
-	action = "WithError()"
-	p.WithError(nil)
-	action = "WithReader()"
-	p.WithReader(strings.NewReader(""))
-	action = "WriteFile()"
-	p.WriteFile(t.TempDir() + "bogus.txt")
-}
-
-func TestNilPipes(t *testing.T) {
-	t.Parallel()
-	doMethodsOnPipe(t, nil, "nil")
-}
-
-func TestZeroPipes(t *testing.T) {
-	t.Parallel()
-	doMethodsOnPipe(t, &script.Pipe{}, "zero")
-}
-
-func TestNewPipes(t *testing.T) {
-	t.Parallel()
-	doMethodsOnPipe(t, script.NewPipe(), "new")
-}
-
-func TestPipeIsReader(t *testing.T) {
-	t.Parallel()
-	var p io.Reader = script.NewPipe()
-	_, err := ioutil.ReadAll(p)
+func TestReadProducesCompletePipeContents(t *testing.T) {
+	t.Parallel()
+	want := []byte("hello")
+	p := script.Echo("hello")
+	got, err := io.ReadAll(p)
 	if err != nil {
-		t.Error(err)
+		t.Fatal(err)
+	}
+	if !cmp.Equal(want, got) {
+		t.Error(cmp.Diff(want, got))
+	}
+}
+
+func TestReadReturnsEOFOnUninitialisedPipe(t *testing.T) {
+	t.Parallel()
+	p := &script.Pipe{}
+	buf := []byte{0} // try to read at least 1 byte
+	n, err := p.Read(buf)
+	if !errors.Is(err, io.EOF) {
+		t.Errorf("want EOF, got %v", err)
+	}
+	if n > 0 {
+		t.Errorf("unexpectedly read %d bytes", n)
 	}
 }
 
@@ -1554,6 +1521,45 @@ func ExamplePipe_First() {
 	// b
 }
 
+func ExamplePipe_Filter() {
+	script.Echo("hello world").Filter(func(r io.Reader, w io.Writer) error {
+		n, err := io.Copy(w, r)
+		fmt.Fprintf(w, "\nfiltered %d bytes\n", n)
+		return err
+	}).Stdout()
+	// Output:
+	// hello world
+	// filtered 11 bytes
+}
+
+func ExamplePipe_FilterScan() {
+	script.Echo("a\nb\nc").FilterScan(func(line string, w io.Writer) {
+		fmt.Fprintf(w, "scanned line: %q\n", line)
+	}).Stdout()
+	// Output:
+	// scanned line: "a"
+	// scanned line: "b"
+	// scanned line: "c"
+}
+
+func ExamplePipe_FilterLine_user() {
+	script.Echo("a\nb\nc").FilterLine(func(line string) string {
+		return "> " + line
+	}).Stdout()
+	// Output:
+	// > a
+	// > b
+	// > c
+}
+
+func ExamplePipe_FilterLine_stdlib() {
+	script.Echo("a\nb\nc").FilterLine(strings.ToUpper).Stdout()
+	// Output:
+	// A
+	// B
+	// C
+}
+
 func ExamplePipe_Freq() {
 	input := strings.Join([]string{
 		"apple",
diff --git a/script_unix_test.go b/script_unix_test.go
index 6a05133..73e7cba 100644
--- a/script_unix_test.go
+++ b/script_unix_test.go
@@ -44,6 +44,7 @@ func TestExecRunsShWithinShWithEchoInceptionAndGetsOutputInception(t *testing.T)
 func TestExecErrorsRunningShellCommandWithUnterminatedStringArgument(t *testing.T) {
 	t.Parallel()
 	p := script.Exec("sh -c 'echo oh no")
+	p.Wait()
 	if p.Error() == nil {
 		t.Error("want error running 'sh' command line containing unterminated string")
 	}