Repository: fsnotify/fsnotify
Pull request #629 (2024-04-28 17:29:04)

Support filtering events, add unsupported events
This adds the ability to only listen for some event types. If you're only interested in Created events and you're getting a lot of Write events then you're just wasting CPU cycles

This also adds the ability listen on extra unportable event types; since this is so related I figured I might as well do both. Ideally we want to 1) make it very very obvious you're doing something unportable, and 2) make it reasonably easy "fallback" for platforms where this isn't supported.

Unportable events start with "Unportable", which should document their unportabilitiness. Also add a new Supports(Op) method, which should make adding fallback logic relatively painless.

For example, to use CloseWrite where supported, but falling back to Write when it's not:

	var op fsnotify.Op
	if w.Supports(fsnotify.UnportableCloseWrite) {
		op |= fsnotify.UnportableCloseWrite
	} else {
		op |= fsnotify.Create | fsnotify.Write
	}
	w.AddWith("/tmp", fsnotify.WithOp(op))

And then you can deal with this in the write loop. There's a full example in cmd/fsnotify/closewrite.go

TODO: need to write tests for this, update all platforms.

Updates #7
Updates #519

nickajacks1: Nice, Supports was a great idea because it lets the programmer explicitly decide what the fallback behavior is. While I'd personally prefer compile time checks, this explicit runtime check method is probably more approachable for people and more in the spirit of fsnotify. A message warning about Unsupported* events in the Readme could also help to make sure devs are aware of any potential misuse.
arp242: > While I'd personally prefer compile time checks, this explicit runtime check method is probably more approachable for people and more in the spirit of fsnotify. 

In principle I agree, however looking at this for a bit, one of my main issues with build tags is that authors will have to make two files like so:

has_closewrite.go:

    //go:build linux || freebsd

    var fsOp = fsnotify.UnportableCloseWrite

    func procEvent(e fsnotify.Event) {
        if e.Has(fsnotify.UnportableCloseWrite) {
            fmt.Println(e)
        }
    }

no_closewrite.go

    //go:build !linux && !freebsd

    var fsOp = fsnotify.Create | fsnotify.Write

    func procEvent(e fsnotify.Event) {
        // Fallback with timers as per ./cmd/fsnotify/dedup.go
    }

And this wouldn't be a huge deal, but those build tags aren't static, and platforms can add support for things like CloseWrite in the future, or fsnotify can add support for things that we didn't have before.

This is not hypothetical: NetBSD 10 (released last month) adds support for CloseWrite and a few other events. Maybe OpenBSD will add it next month or next year? Entirely possible. When/if we add support for fanotify, fsevents, polling, the feature sets for those backends will be different (but the GOOS build tag won't be).

So a 1:1 mapping of operating system to supported features is neither static nor perfect.

And it's not an uncommon situation for someone to be developing on, say, macOS, having a CI for Windows and Linux, and BSD systems will more or less "just work" automatically. By focusing on platforms (go:build linux) rather than features (w.Supports(fsnotify.UnportableCloseWrite)) things will probably break more on lesser-used systems, or use sub-optimal solutions on those systems.

All things considered, runtime checks are probably the right choice here, and prefixing the name with UnportableEvent should give enough of a clue that it's unportable, and people will have to make a conscious decision about this.

---

The main downside of Watcher.Supports is that it's tied to the Watcher. When I added AddWith() to pass options in #521, I imagined it could be used to select the backend, among other things; e.g.

    w.AddWith("/path", fsnotify.WithFanotify())
    w.AddWith("/path", fsnotify.WithPoll())

But that won't jibe with Watcher.Supports(). I'm okay with just telling people to use different Watcher instances for different backends, as that's probably quite a rare use case. But need to be able to override the default backend selection. Having a method for that is probably okay:

    func (w *Watcher) Backend(b Backend) error {
        if w.startedWatching {
            return errors.New("already started")
        }
        w.backend = b
        return nil
    }

Adding options to NewWatcher() was discussed in #550 when NewBufferedWatcher() was added.

Patch:
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
index 5a31be0..cc97b75 100644
--- a/CONTRIBUTING.md
+++ b/CONTRIBUTING.md
@@ -67,8 +67,9 @@ End-of-line escapes with `\` are not supported.
 
 ### Supported commands
 
-    watch path          # Watch the path, reporting events for it. Nothing is
-                        # watched by default.
+    watch path [ops]    # Watch the path, reporting events for it. Nothing is
+                        # watched by default. Optionally a list of ops can be
+                        # given, as with AddWith(path, WithOps(...)).
     unwatch path        # Stop watching the path.
     watchlist n         # Assert watchlist length.
 
@@ -87,6 +88,7 @@ End-of-line escapes with `\` are not supported.
     chmod mode path     # Octal only
     sleep time-in-ms
 
+    cat path            # Read path (does nothing with the data; just reads it).
     echo str >>path     # Append "str" to "path".
     echo str >path      # Truncate "path" and write "str".
 
diff --git a/backend_fen.go b/backend_fen.go
index 662ffbc..a816760 100644
--- a/backend_fen.go
+++ b/backend_fen.go
@@ -131,9 +131,9 @@ type Watcher struct {
 
 	mu      sync.Mutex
 	port    *unix.EventPort
-	done    chan struct{}       // Channel for sending a "quit message" to the reader goroutine
-	dirs    map[string]struct{} // Explicitly watched directories
-	watches map[string]struct{} // Explicitly watched non-directories
+	done    chan struct{} // Channel for sending a "quit message" to the reader goroutine
+	dirs    map[string]Op // Explicitly watched directories
+	watches map[string]Op // Explicitly watched non-directories
 }
 
 // NewWatcher creates a new Watcher.
@@ -153,8 +153,8 @@ func NewBufferedWatcher(sz uint) (*Watcher, error) {
 	w := &Watcher{
 		Events:  make(chan Event, sz),
 		Errors:  make(chan error),
-		dirs:    make(map[string]struct{}),
-		watches: make(map[string]struct{}),
+		dirs:    make(map[string]Op),
+		watches: make(map[string]Op),
 		done:    make(chan struct{}),
 	}
 
@@ -269,7 +269,10 @@ func (w *Watcher) AddWith(name string, opts ...addOpt) error {
 			time.Now().Format("15:04:05.000000000"), name)
 	}
 
-	_ = getOptions(opts...)
+	with := getOptions(opts...)
+	if !w.xSupports(with.op) {
+		return fmt.Errorf("%w: %s", xErrUnsupported, with.op)
+	}
 
 	// Currently we resolve symlinks that were explicitly requested to be
 	// watched. Otherwise we would use LStat here.
@@ -286,7 +289,7 @@ func (w *Watcher) AddWith(name string, opts ...addOpt) error {
 		}
 
 		w.mu.Lock()
-		w.dirs[name] = struct{}{}
+		w.dirs[name] = with.op
 		w.mu.Unlock()
 		return nil
 	}
@@ -297,7 +300,7 @@ func (w *Watcher) AddWith(name string, opts ...addOpt) error {
 	}
 
 	w.mu.Lock()
-	w.watches[name] = struct{}{}
+	w.watches[name] = with.op
 	w.mu.Unlock()
 	return nil
 }
@@ -594,20 +597,24 @@ func (w *Watcher) associateFile(path string, stat os.FileInfo, follow bool) erro
 		// cleared up that discrepancy. The most likely cause is that the event
 		// has fired but we haven't processed it yet.
 		err := w.port.DissociatePath(path)
-		if err != nil && err != unix.ENOENT {
+		if err != nil && !errors.Is(err, unix.ENOENT) {
 			return err
 		}
 	}
-	// FILE_NOFOLLOW means we watch symlinks themselves rather than their
-	// targets.
-	events := unix.FILE_MODIFIED | unix.FILE_ATTRIB | unix.FILE_NOFOLLOW
-	if follow {
-		// We *DO* follow symlinks for explicitly watched entries.
-		events = unix.FILE_MODIFIED | unix.FILE_ATTRIB
+
+	var events int
+	if !follow {
+		// Watch symlinks themselves rather than their targets unless this entry
+		// is explicitly watched.
+		events |= unix.FILE_NOFOLLOW
+	}
+	if true { // TODO: implement withOps()
+		events |= unix.FILE_MODIFIED
 	}
-	return w.port.AssociatePath(path, stat,
-		events,
-		stat.Mode())
+	if true {
+		events |= unix.FILE_ATTRIB
+	}
+	return w.port.AssociatePath(path, stat, events, stat.Mode())
 }
 
 func (w *Watcher) dissociateFile(path string, stat os.FileInfo, unused bool) error {
@@ -639,3 +646,15 @@ func (w *Watcher) WatchList() []string {
 
 	return entries
 }
+
+// Supports reports if all the listed operations are supported by this platform.
+//
+// Create, Write, Remove, Rename, and Chmod are always supported. It can only
+// return false for an Op starting with Unportable.
+func (w *Watcher) xSupports(op Op) bool {
+	if op.Has(xUnportableOpen) || op.Has(xUnportableRead) ||
+		op.Has(xUnportableCloseWrite) || op.Has(xUnportableCloseRead) {
+		return false
+	}
+	return true
+}
diff --git a/backend_inotify.go b/backend_inotify.go
index 2937197..3d79754 100644
--- a/backend_inotify.go
+++ b/backend_inotify.go
@@ -397,13 +397,41 @@ func (w *Watcher) AddWith(name string, opts ...addOpt) error {
 			time.Now().Format("15:04:05.000000000"), name)
 	}
 
-	name = filepath.Clean(name)
-	_ = getOptions(opts...)
+	with := getOptions(opts...)
+	if !w.xSupports(with.op) {
+		return fmt.Errorf("%w: %s", xErrUnsupported, with.op)
+	}
 
-	var flags uint32 = unix.IN_MOVED_TO | unix.IN_MOVED_FROM |
-		unix.IN_CREATE | unix.IN_ATTRIB | unix.IN_MODIFY |
-		unix.IN_MOVE_SELF | unix.IN_DELETE | unix.IN_DELETE_SELF
+	var flags uint32
+	if with.op.Has(Create) {
+		flags |= unix.IN_CREATE
+	}
+	if with.op.Has(Write) {
+		flags |= unix.IN_MODIFY
+	}
+	if with.op.Has(Remove) {
+		flags |= unix.IN_DELETE | unix.IN_DELETE_SELF
+	}
+	if with.op.Has(Rename) {
+		flags |= unix.IN_MOVED_TO | unix.IN_MOVED_FROM | unix.IN_MOVE_SELF
+	}
+	if with.op.Has(Chmod) {
+		flags |= unix.IN_ATTRIB
+	}
+	if with.op.Has(xUnportableOpen) {
+		flags |= unix.IN_OPEN
+	}
+	if with.op.Has(xUnportableRead) {
+		flags |= unix.IN_ACCESS
+	}
+	if with.op.Has(xUnportableCloseWrite) {
+		flags |= unix.IN_CLOSE_WRITE
+	}
+	if with.op.Has(xUnportableCloseRead) {
+		flags |= unix.IN_CLOSE_NOWRITE
+	}
 
+	name = filepath.Clean(name)
 	return w.watches.updatePath(name, func(existing *watch) (*watch, error) {
 		if existing != nil {
 			flags |= existing.flags | unix.IN_MASK_ADD
@@ -624,6 +652,18 @@ func (w *Watcher) newEvent(name string, mask, cookie uint32) Event {
 	if mask&unix.IN_MODIFY == unix.IN_MODIFY {
 		e.Op |= Write
 	}
+	if mask&unix.IN_OPEN == unix.IN_OPEN {
+		e.Op |= xUnportableOpen
+	}
+	if mask&unix.IN_ACCESS == unix.IN_ACCESS {
+		e.Op |= xUnportableRead
+	}
+	if mask&unix.IN_CLOSE_WRITE == unix.IN_CLOSE_WRITE {
+		e.Op |= xUnportableCloseWrite
+	}
+	if mask&unix.IN_CLOSE_NOWRITE == unix.IN_CLOSE_NOWRITE {
+		e.Op |= xUnportableCloseRead
+	}
 	if mask&unix.IN_MOVE_SELF == unix.IN_MOVE_SELF || mask&unix.IN_MOVED_FROM == unix.IN_MOVED_FROM {
 		e.Op |= Rename
 	}
@@ -655,3 +695,11 @@ func (w *Watcher) newEvent(name string, mask, cookie uint32) Event {
 	}
 	return e
 }
+
+// Supports reports if all the listed operations are supported by this platform.
+//
+// Create, Write, Remove, Rename, and Chmod are always supported. It can only
+// return false for an Op starting with Unportable.
+func (w *Watcher) xSupports(op Op) bool {
+	return true // Supports everything.
+}
diff --git a/backend_kqueue.go b/backend_kqueue.go
index 2fd5796..9e6aa9d 100644
--- a/backend_kqueue.go
+++ b/backend_kqueue.go
@@ -458,7 +458,10 @@ func (w *Watcher) AddWith(name string, opts ...addOpt) error {
 			time.Now().Format("15:04:05.000000000"), name)
 	}
 
-	_ = getOptions(opts...)
+	with := getOptions(opts...)
+	if !w.xSupports(with.op) {
+		return fmt.Errorf("%w: %s", xErrUnsupported, with.op)
+	}
 
 	w.watches.addUserWatch(name)
 	_, err := w.addWatch(name, noteAllEvents)
@@ -886,3 +889,18 @@ func (w *Watcher) read(events []unix.Kevent_t) ([]unix.Kevent_t, error) {
 	}
 	return events[0:n], nil
 }
+
+// Supports reports if all the listed operations are supported by this platform.
+//
+// Create, Write, Remove, Rename, and Chmod are always supported. It can only
+// return false for an Op starting with Unportable.
+func (w *Watcher) xSupports(op Op) bool {
+	if runtime.GOOS == "freebsd" {
+		//return true // Supports everything.
+	}
+	if op.Has(xUnportableOpen) || op.Has(xUnportableRead) ||
+		op.Has(xUnportableCloseWrite) || op.Has(xUnportableCloseRead) {
+		return false
+	}
+	return true
+}
diff --git a/backend_other.go b/backend_other.go
index f868fca..9cf8276 100644
--- a/backend_other.go
+++ b/backend_other.go
@@ -196,3 +196,9 @@ func (w *Watcher) AddWith(name string, opts ...addOpt) error { return nil }
 //
 // Returns nil if [Watcher.Close] was called.
 func (w *Watcher) Remove(name string) error { return nil }
+
+// Supports reports if all the listed operations are supported by this platform.
+//
+// Create, Write, Remove, Rename, and Chmod are always supported. It can only
+// return false for an Op starting with Unportable.
+func (w *Watcher) xSupports(op Op) bool { return false }
diff --git a/backend_windows.go b/backend_windows.go
index 564f52d..688e698 100644
--- a/backend_windows.go
+++ b/backend_windows.go
@@ -280,6 +280,9 @@ func (w *Watcher) AddWith(name string, opts ...addOpt) error {
 	}
 
 	with := getOptions(opts...)
+	if !w.xSupports(with.op) {
+		return fmt.Errorf("%w: %s", xErrUnsupported, with.op)
+	}
 	if with.bufsize < 4096 {
 		return fmt.Errorf("fsnotify.WithBufferSize: buffer size cannot be smaller than 4096 bytes")
 	}
@@ -841,3 +844,15 @@ func (w *Watcher) toFSnotifyFlags(action uint32) uint64 {
 	}
 	return 0
 }
+
+// Supports reports if all the listed operations are supported by this platform.
+//
+// Create, Write, Remove, Rename, and Chmod are always supported. It can only
+// return false for an Op starting with Unportable.
+func (w *Watcher) xSupports(op Op) bool {
+	if op.Has(xUnportableOpen) || op.Has(xUnportableRead) ||
+		op.Has(xUnportableCloseWrite) || op.Has(xUnportableCloseRead) {
+		return false
+	}
+	return true
+}
diff --git a/cmd/fsnotify/closewrite.go b/cmd/fsnotify/closewrite.go
new file mode 100644
index 0000000..ecbc90c
--- /dev/null
+++ b/cmd/fsnotify/closewrite.go
@@ -0,0 +1,89 @@
+package main
+
+/*
+func closeWrite(paths ...string) {
+	if len(paths) < 1 {
+		exit("must specify at least one path to watch")
+	}
+
+	w, err := fsnotify.NewWatcher()
+	if err != nil {
+		exit("creating a new watcher: %s", err)
+	}
+	defer w.Close()
+
+	var (
+		op fsnotify.Op
+		cw = w.Supports(fsnotify.UnportableCloseWrite)
+	)
+	if cw {
+		op |= fsnotify.UnportableCloseWrite
+	} else {
+		op |= fsnotify.Create | fsnotify.Write
+	}
+
+	go closeWriteLoop(w, cw)
+
+	for _, p := range paths {
+		err := w.AddWith(p, fsnotify.WithOps(op))
+		if err != nil {
+			exit("%q: %s", p, err)
+		}
+	}
+
+	printTime("ready; press ^C to exit")
+	<-make(chan struct{})
+}
+
+func closeWriteLoop(w *fsnotify.Watcher, cw bool) {
+	var (
+		waitFor = 100 * time.Millisecond
+		mu      sync.Mutex
+		timers  = make(map[string]*time.Timer)
+	)
+	for {
+		select {
+		case err, ok := <-w.Errors:
+			if !ok {
+				return
+			}
+			panic(err)
+		case e, ok := <-w.Events:
+			if !ok {
+				return
+			}
+
+			// CloseWrite is supported: easy case.
+			if cw {
+				if e.Has(fsnotify.UnportableCloseWrite) {
+					printTime(e.String())
+				}
+				continue
+			}
+
+			// Get timer.
+			mu.Lock()
+			t, ok := timers[e.Name]
+			mu.Unlock()
+
+			// No timer yet, so create one.
+			if !ok {
+				t = time.AfterFunc(math.MaxInt64, func() {
+					printTime(e.String())
+					mu.Lock()
+					delete(timers, e.Name)
+					mu.Unlock()
+				})
+				t.Stop()
+
+				mu.Lock()
+				timers[e.Name] = t
+				mu.Unlock()
+			}
+
+			// Reset the timer for this path, so it will start from 100ms again.
+			t.Reset(waitFor)
+		}
+	}
+}
+*/
diff --git a/fsnotify.go b/fsnotify.go
index 21bf887..b43d524 100644
--- a/fsnotify.go
+++ b/fsnotify.go
@@ -86,6 +86,31 @@ const (
 	// get triggered very frequently by some software. For example, Spotlight
 	// indexing on macOS, anti-virus software, backup software, etc.
 	Chmod
+
+	// File descriptor was opened.
+	//
+	// Only works on Linux and FreeBSD.
+	xUnportableOpen
+
+	// File was read from.
+	//
+	// Only works on Linux and FreeBSD.
+	xUnportableRead
+
+	// File opened for writing was closed.
+	//
+	// Only works on Linux and FreeBSD.
+	//
+	// The advantage of using this over Write is that it's more reliable than
+	// waiting for Write events to stop. It's also faster (if you're not
+	// listening to Write events): copying a file of a few GB can easily
+	// generate tens of thousands of Write events in a short span of time.
+	xUnportableCloseWrite
+
+	// File opened for reading was closed.
+	//
+	// Only works on Linux and FreeBSD.
+	xUnportableCloseRead
 )
 
 var (
@@ -105,6 +130,10 @@ var (
 	//  - windows:      The buffer size is too small; WithBufferSize() can be used to increase it.
 	//  - kqueue, fen:  Not used.
 	ErrEventOverflow = errors.New("fsnotify: queue or buffer overflow")
+
+	// ErrUnsupported is returned by AddWith() when WithOps() specified an
+	// Unportable event that's not supported on this platform.
+	xErrUnsupported = errors.New("fsnotify: not supported with this backend")
 )
 
 func (o Op) String() string {
@@ -118,6 +147,18 @@ func (o Op) String() string {
 	if o.Has(Write) {
 		b.WriteString("|WRITE")
 	}
+	if o.Has(xUnportableOpen) {
+		b.WriteString("|OPEN")
+	}
+	if o.Has(xUnportableRead) {
+		b.WriteString("|READ")
+	}
+	if o.Has(xUnportableCloseWrite) {
+		b.WriteString("|CLOSE_WRITE")
+	}
+	if o.Has(xUnportableCloseRead) {
+		b.WriteString("|CLOSE_READ")
+	}
 	if o.Has(Rename) {
 		b.WriteString("|RENAME")
 	}
@@ -148,6 +189,7 @@ type (
 	addOpt   func(opt *withOpts)
 	withOpts struct {
 		bufsize int
+		op      Op
 	}
 )
 
@@ -160,6 +202,7 @@ var debug = func() bool {
 
 var defaultOpts = withOpts{
 	bufsize: 65536, // 64K
+	op:      Create | Write | Remove | Rename | Chmod,
 }
 
 func getOptions(opts ...addOpt) withOpts {
@@ -184,6 +227,24 @@ func WithBufferSize(bytes int) addOpt {
 	return func(opt *withOpts) { opt.bufsize = bytes }
 }
 
+// WithOps sets which operations to listen for. The default is [Create],
+// [Write], [Remove], [Rename], and [Chmod].
+//
+// Excluding operations you're not interested in can save quite a bit of CPU
+// time; in some use cases there may be hundreds of thousands of useless Write
+// or Chmod operations per second.
+//
+// This can also be used to add unportable operations not supported by all
+// platforms; unportable operations all start with "Unportable":
+// [UnportableOpen], [UnportableRead], [UnportableCloseWrite], and
+// [UnportableCloseRead].
+//
+// AddWith returns an error when using an unportable operation that's not
+// supported. Use [Watcher.Support] to check for support.
+func withOps(op Op) addOpt {
+	return func(opt *withOpts) { opt.op = op }
+}
+
 var enableRecurse = false
 
 // Check if this path is recursive (ends with "/..." or "\..."), and return the
diff --git a/mkdoc.zsh b/mkdoc.zsh
index e3c1caf..6e630ec 100755
--- a/mkdoc.zsh
+++ b/mkdoc.zsh
@@ -167,6 +167,14 @@ watchlist=$(<<EOF
 EOF
 )
 
+supports=$(<<EOF
+// Supports reports if all the listed operations are supported by this platform.
+//
+// Create, Write, Remove, Rename, and Chmod are always supported. It can only
+// return false for an Op starting with Unportable.
+EOF
+)
+
 events=$(<<EOF
 	// Events sends the filesystem change events.
 	//
@@ -249,5 +257,6 @@ set-cmt '^func (w \*Watcher) AddWith('      $addwith
 set-cmt '^func (w \*Watcher) Remove('       $remove
 set-cmt '^func (w \*Watcher) Close('        $close
 set-cmt '^func (w \*Watcher) WatchList('    $watchlist
+set-cmt '^func (w \*Watcher) Supports('     $supports
 set-cmt '^[[:space:]]*Events *chan Event$'  $events
 set-cmt '^[[:space:]]*Errors *chan error$'  $errors


Test patch:
diff --git a/fsnotify_test.go b/fsnotify_test.go
index ac7d142..08e049f 100644
--- a/fsnotify_test.go
+++ b/fsnotify_test.go
@@ -599,7 +599,7 @@ func TestRemove(t *testing.T) {
 	})
 
 	t.Run("remove with ... when non-recursive", func(t *testing.T) {
-		recurseOnly(t)
+		supportsRecurse(t)
 		t.Parallel()
 
 		tmp := t.TempDir()
diff --git a/helpers_test.go b/helpers_test.go
index 5c74e3a..cc46c27 100644
--- a/helpers_test.go
+++ b/helpers_test.go
@@ -331,6 +331,21 @@ func rmAll(t *testing.T, path ...string) {
 	}
 }
 
+// cat
+func cat(t *testing.T, path ...string) {
+	t.Helper()
+	if len(path) < 1 {
+		t.Fatalf("cat: path must have at least one element: %s", path)
+	}
+	_, err := os.ReadFile(join(path...))
+	if err != nil {
+		t.Fatalf("cat(%q): %s", join(path...), err)
+	}
+	if shouldWait(path...) {
+		eventSeparator()
+	}
+}
+
 // chmod
 func chmod(t *testing.T, mode fs.FileMode, path ...string) {
 	t.Helper()
@@ -539,6 +554,14 @@ func newEvents(t *testing.T, s string) Events {
 				op |= Rename
 			case "CHMOD":
 				op |= Chmod
+			case "OPEN":
+				op |= xUnportableOpen
+			case "READ":
+				op |= xUnportableRead
+			case "CLOSE_WRITE":
+				op |= xUnportableCloseWrite
+			case "CLOSE_READ":
+				op |= xUnportableCloseRead
 			default:
 				t.Fatalf("newEvents: line %d has unknown event %q: %s", no+1, ee, line)
 			}
@@ -603,11 +626,6 @@ func indent(s fmt.Stringer) string {
 
 var join = filepath.Join
 
-func isCI() bool {
-	_, ok := os.LookupEnv("CI")
-	return ok
-}
-
 func isKqueue() bool {
 	switch runtime.GOOS {
 	case "darwin", "freebsd", "openbsd", "netbsd", "dragonfly":
@@ -624,7 +642,7 @@ func isSolaris() bool {
 	return false
 }
 
-func recurseOnly(t *testing.T) {
+func supportsRecurse(t *testing.T) {
 	switch runtime.GOOS {
 	case "windows":
 		// Run test.
@@ -633,6 +651,15 @@ func recurseOnly(t *testing.T) {
 	}
 }
 
+func supportsFilter(t *testing.T) {
+	switch runtime.GOOS {
+	case "linux":
+		// Run test.
+	default:
+		t.Skip("withOps() not yet supported on " + runtime.GOOS)
+	}
+}
+
 func supportsRename() bool {
 	switch runtime.GOOS {
 	case "linux", "windows":
@@ -738,6 +765,26 @@ loop:
 		case "skip", "require":
 			mustArg(c, 1)
 			switch c.args[0] {
+			case "op_all":
+				if runtime.GOOS != "linux" {
+					t.Skip("No op_all on this platform")
+				}
+			case "op_open":
+				if runtime.GOOS != "linux" {
+					t.Skip("No Open on this platform")
+				}
+			case "op_read":
+				if runtime.GOOS != "linux" {
+					t.Skip("No Read on this platform")
+				}
+			case "op_close_write":
+				if runtime.GOOS != "linux" {
+					t.Skip("No CloseWrite on this platform")
+				}
+			case "op_close_read":
+				if runtime.GOOS != "linux" {
+					t.Skip("No CloseRead on this platform")
+				}
 			case "always":
 				t.Skip()
 			case "symlink":
@@ -762,7 +809,9 @@ loop:
 					t.Skip(`"mknod fails with "not owner"`)
 				}
 			case "recurse":
-				recurseOnly(t)
+				supportsRecurse(t)
+			case "filter":
+				supportsFilter(t)
 			case "windows":
 				if runtime.GOOS == "windows" {
 					t.Skip("Skipping on Windows")
@@ -771,6 +820,10 @@ loop:
 				if runtime.GOOS == "netbsd" {
 					t.Skip("Skipping on NetBSD")
 				}
+			case "openbsd":
+				if runtime.GOOS == "openbsd" {
+					t.Skip("Skipping on OpenBSD")
+				}
 			default:
 				t.Fatalf("line %d: unknown %s reason: %q", c.line, c.cmd, c.args[0])
 			}
@@ -791,8 +844,49 @@ loop:
 			mustArg(c, 0)
 			break loop
 		case "watch":
-			mustArg(c, 1)
-			do = append(do, func() { addWatch(t, w.w, tmppath(tmp, c.args[0])) })
+			if len(c.args) < 1 {
+				t.Fatalf("line %d: %q requires at least %d arguments (have %d: %q)",
+					c.line, c.cmd, 1, len(c.args), c.args)
+			}
+			if len(c.args) == 1 {
+				do = append(do, func() { addWatch(t, w.w, tmppath(tmp, c.args[0])) })
+				continue
+			}
+			var op Op
+			for _, o := range c.args[1:] {
+				switch strings.ToLower(o) {
+				default:
+					t.Fatalf("line %d: unknown: %q", c.line+1, o)
+				case "default":
+					op |= Create | Write | Remove | Rename | Chmod
+				case "create":
+					op |= Create
+				case "write":
+					op |= Write
+				case "remove":
+					op |= Remove
+				case "rename":
+					op |= Rename
+				case "chmod":
+					op |= Chmod
+				case "open":
+					op |= xUnportableOpen
+				case "read":
+					op |= xUnportableRead
+				case "close_write":
+					op |= xUnportableCloseWrite
+				case "close_read":
+					op |= xUnportableCloseRead
+				}
+
+			}
+			do = append(do, func() {
+				p := tmppath(tmp, c.args[0])
+				err := w.w.AddWith(p, withOps(op))
+				if err != nil {
+					t.Fatalf("line %d: addWatch(%q): %s", c.line+1, p, err)
+				}
+			})
 		case "unwatch":
 			mustArg(c, 1)
 			do = append(do, func() { rmWatch(t, w.w, tmppath(tmp, c.args[0])) })
@@ -859,6 +953,9 @@ loop:
 				t.Fatalf("line %d: %s", c.line, err)
 			}
 			do = append(do, func() { chmod(t, fs.FileMode(n), tmppath(tmp, c.args[1])) })
+		case "cat":
+			mustArg(c, 1)
+			do = append(do, func() { cat(t, tmppath(tmp, c.args[0])) })
 		case "echo":
 			if len(c.args) < 2 || len(c.args) > 3 {
 				t.Fatalf("line %d: %q requires 2 or 3 arguments (have %d: %q)",
diff --git a/testdata/watch-dir/multiple-creates b/testdata/watch-dir/multiple-creates
index bec1615..1f89f88 100644
--- a/testdata/watch-dir/multiple-creates
+++ b/testdata/watch-dir/multiple-creates
@@ -1,5 +1,9 @@
 # Create a file, delete it, and re-create the same file again.
 
+# TODO: this started failing in the CI, and I don't really know why?
+skip netbsd
+skip openbsd
+
 watch /
 
 echo data >>/file
@@ -10,9 +14,9 @@ echo data >>/file  # Modify
 echo data >>/file  # Modify
 
 Output:
-	create  /file
-	write   /file
-	remove  /file
-	create  /file
-	write   /file
+	create  /file  # echo data >>/file
 	write   /file
+	remove  /file  # rm /file
+	create  /file  # touch /file
+	write   /file  # echo data >>/file
+	write   /file  # echo data >>/file
diff --git a/testdata/watch-dir/only-chmod b/testdata/watch-dir/only-chmod
new file mode 100644
index 0000000..f108505
--- /dev/null
+++ b/testdata/watch-dir/only-chmod
@@ -0,0 +1,36 @@
+# Listen for chmod events only.
+require filter
+watch /   chmod
+
+# Create
+touch /file
+mkdir /dir
+ln -s /file /link
+
+# Write
+echo data >>/file
+echo data >>/link
+touch /dir/file
+
+# Rename
+mv /file /rename
+mv /rename /file
+mv /dir /rename
+mv /rename /dir
+mv /link /rename
+mv /rename /link
+
+# Chmod
+chmod 644 /file
+chmod 644 /link
+chmod 755 /dir
+
+# Remove
+rm /file
+rm /link
+rm -r /dir
+
+Output:
+	chmod /file
+	chmod /file
+	chmod /dir
diff --git a/testdata/watch-dir/only-create b/testdata/watch-dir/only-create
new file mode 100644
index 0000000..8c289be
--- /dev/null
+++ b/testdata/watch-dir/only-create
@@ -0,0 +1,36 @@
+# Listen for create events only.
+require filter
+watch /   create
+
+# Create
+touch /file
+mkdir /dir
+ln -s /file /link
+
+# Write
+echo data >>/file
+echo data >>/link
+touch /dir/file
+
+# Rename
+mv /file /rename
+mv /rename /file
+mv /dir /rename
+mv /rename /dir
+mv /link /rename
+mv /rename /link
+
+# Chmod
+chmod 644 /file
+chmod 644 /link
+chmod 755 /dir
+
+# Remove
+rm /file
+rm /link
+rm -r /dir
+
+Output:
+	create  /file
+	create  /link
+	create  /dir
diff --git a/testdata/watch-dir/only-remove b/testdata/watch-dir/only-remove
new file mode 100644
index 0000000..8f7a6fe
--- /dev/null
+++ b/testdata/watch-dir/only-remove
@@ -0,0 +1,36 @@
+# Listen for remove events only.
+require filter
+watch /   remove
+
+# Create
+touch /file
+mkdir /dir
+ln -s /file /link
+
+# Write
+echo data >>/file
+echo data >>/link
+touch /dir/file
+
+# Rename
+mv /file /rename
+mv /rename /file
+mv /dir /rename
+mv /rename /dir
+mv /link /rename
+mv /rename /link
+
+# Chmod
+chmod 644 /file
+chmod 644 /link
+chmod 755 /dir
+
+# Remove
+rm /file
+rm /link
+rm -r /dir
+
+Output:
+	remove  /file
+	remove  /link
+	remove  /dir
diff --git a/testdata/watch-dir/only-rename b/testdata/watch-dir/only-rename
new file mode 100644
index 0000000..4343657
--- /dev/null
+++ b/testdata/watch-dir/only-rename
@@ -0,0 +1,50 @@
+# Listen for rename events only.
+require filter
+watch /   rename
+
+# Create
+touch /file
+mkdir /dir
+ln -s /file /link
+
+# Write
+echo data >>/file
+echo data >>/link
+touch /dir/file
+
+# Rename
+mv /file /rename
+mv /rename /file
+mv /dir /rename
+mv /rename /dir
+mv /link /rename
+mv /rename /link
+
+# Chmod
+chmod 644 /file
+chmod 644 /link
+chmod 755 /dir
+
+# Remove
+rm /file
+rm /link
+rm -r /dir
+
+Output:
+	rename   /file
+	create   /rename ← /file
+
+	rename   /rename
+	create   /file ← /rename
+
+	rename   /dir
+	create   /rename ← /dir
+
+	rename   /rename
+	create   /dir ← /rename
+
+	rename   /link
+	create   /rename ← /link
+
+	rename   /rename
+	create   /link ← /rename
diff --git a/testdata/watch-dir/only-write b/testdata/watch-dir/only-write
new file mode 100644
index 0000000..8348ab1
--- /dev/null
+++ b/testdata/watch-dir/only-write
@@ -0,0 +1,35 @@
+# Listen for write events only.
+require filter
+watch /   write
+
+# Create
+touch /file
+mkdir /dir
+ln -s /file /link
+
+# Write
+echo data >>/file
+echo data >>/link
+touch /dir/file
+
+# Rename
+mv /file /rename
+mv /rename /file
+mv /dir /rename
+mv /rename /dir
+mv /link /rename
+mv /rename /link
+
+# Chmod
+chmod 644 /file
+chmod 644 /link
+chmod 755 /dir
+
+# Remove
+rm /file
+rm /link
+rm -r /dir
+
+Output:
+	write /file
+	write /file
diff --git a/testdata/watch-dir/op-all b/testdata/watch-dir/op-all
new file mode 100644
index 0000000..6472c6b
--- /dev/null
+++ b/testdata/watch-dir/op-all
@@ -0,0 +1,38 @@
+require op_all
+
+watch /  default open read close_write close_read
+touch /file
+ln -s /file /link
+
+echo data >>/file2
+
+cat /file
+cat /file2
+
+cat /link
+
+mkdir /dir
+
+Output:
+	create      /file
+	open        /file
+	close_write /file
+
+	create      /link
+
+	create      /file2
+	open        /file2
+	write       /file2
+	close_write /file2
+
+	open        /file
+	close_read  /file
+
+	open        /file2
+	read        /file2
+	close_read  /file2
+
+	open        /file
+	close_read  /file
+
+	create /dir
diff --git a/testdata/watch-dir/op-closeread b/testdata/watch-dir/op-closeread
new file mode 100644
index 0000000..e794dbe
--- /dev/null
+++ b/testdata/watch-dir/op-closeread
@@ -0,0 +1,9 @@
+# Basic close_read test
+require op_close_read
+
+echo "hello, world!" >>/file
+watch /   default close_read
+cat /file
+
+Output:
+	close_read /file
diff --git a/testdata/watch-dir/op-closewrite b/testdata/watch-dir/op-closewrite
new file mode 100644
index 0000000..816c39c
--- /dev/null
+++ b/testdata/watch-dir/op-closewrite
@@ -0,0 +1,20 @@
+# Basic close_write test
+require op_close_write
+
+watch /   default close_write
+
+touch /file
+echo "hello, world!" >>/file
+cat /file
+echo "hello, world!" >/file
+
+Output:
+	create        /file  # touch
+	close_write   /file
+
+	write         /file  # echo >>
+	close_write   /file
+
+	write         /file  # echo >
+	write         /file
+	close_write   /file
diff --git a/testdata/watch-dir/op-open b/testdata/watch-dir/op-open
new file mode 100644
index 0000000..18b24d0
--- /dev/null
+++ b/testdata/watch-dir/op-open
@@ -0,0 +1,9 @@
+# Basic open test
+require op_open
+
+echo "hello, world!" >>/file
+watch /   default open
+cat /file
+
+Output:
+	open /file
diff --git a/testdata/watch-dir/op-read b/testdata/watch-dir/op-read
new file mode 100644
index 0000000..e5a87ac
--- /dev/null
+++ b/testdata/watch-dir/op-read
@@ -0,0 +1,9 @@
+# Basic close_read test
+require op_read
+
+echo "hello, world!" >>/file
+watch /   default read
+cat /file
+
+Output:
+	read /file
diff --git a/testdata/watch-file/op-all b/testdata/watch-file/op-all
new file mode 100644
index 0000000..22f327e
--- /dev/null
+++ b/testdata/watch-file/op-all
@@ -0,0 +1,20 @@
+require op_all
+
+touch /file
+watch /file  default open read close_write close_read
+cat /file
+
+echo data >>/file
+cat /file
+
+Output:
+	open        /file  # cat /file
+	close_read  /file
+
+	open        /file  # echo data >>/file
+	write       /file
+	close_write /file
+
+	open        /file # cat /file
+	read        /file
+	close_read  /file
diff --git a/testdata/watch-file/op-closeread b/testdata/watch-file/op-closeread
new file mode 100644
index 0000000..52d9928
--- /dev/null
+++ b/testdata/watch-file/op-closeread
@@ -0,0 +1,9 @@
+# Basic close_read test
+require op_close_read
+
+echo "hello, world!" >>/file
+watch /file   default close_read
+cat /file
+
+Output:
+	close_read /file
diff --git a/testdata/watch-file/op-closewrite b/testdata/watch-file/op-closewrite
new file mode 100644
index 0000000..b320b60
--- /dev/null
+++ b/testdata/watch-file/op-closewrite
@@ -0,0 +1,16 @@
+# Basic close_write test
+require op_close_write
+
+touch /file
+watch /file   default close_write
+
+echo "hello, world!" >>/file
+echo "hello, world!" >/file
+
+Output:
+	write         /file  # echo >>
+	close_write   /file
+
+	write         /file  # echo >
+	write         /file
+	close_write   /file
diff --git a/testdata/watch-file/op-open b/testdata/watch-file/op-open
new file mode 100644
index 0000000..2f493c4
--- /dev/null
+++ b/testdata/watch-file/op-open
@@ -0,0 +1,9 @@
+# Basic open test
+require op_open
+
+echo "hello, world!" >>/file
+watch /file   default open
+cat /file
+
+Output:
+	open /file
diff --git a/testdata/watch-file/op-read b/testdata/watch-file/op-read
new file mode 100644
index 0000000..a3df705
--- /dev/null
+++ b/testdata/watch-file/op-read
@@ -0,0 +1,9 @@
+# Basic close_read test
+require op_read
+
+echo "hello, world!" >>/file
+watch /file   default read
+cat /file
+
+Output:
+	read /file