rework watcher tests (#53)

notify.go

@@ -9,7 +9,7 @@ var g *Tree
 func tree() *Tree {
 	once.Do(func() {
 		if g == nil {
-			g = NewTree(NewWatcher())
+			g = NewTree(NewWatcher(nil))
 		}
 	})
 	return g

sync_unix_test.go

@@ -0,0 +1,10 @@
+// +build !windows
+
+package notify
+
+import "golang.org/x/sys/unix"
+
+// TODO
+func Sync() {
+	unix.Sync()
+}

sync_windows_test.go

@@ -0,0 +1,16 @@
+// +build windows
+
+package notify
+
+import "golang.org/x/sys/windows"
+
+var modkernel32 = windows.NewLazyDLL("kernel32.dll")
+var procSetSystemFileCacheSize = modkernel32.NewProc("SetSystemFileCacheSize")
+
+// TODO
+func Sync() {
+	r, _, err := procSetSystemFileCacheSize.Call(-1, -1, 0)
+	if r == 0 {
+		panic(err)
+	}
+}

test_test.go

@@ -0,0 +1,215 @@
+package notify
+
+import (
+	"bufio"
+	"fmt"
+	"io/ioutil"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+)
+
+// TODO(rjeczalik): add debug printing
+
+func isDir(path string) bool {
+	r := path[len(path)-1]
+	return r == '\\' || r == '/'
+}
+
+func tmpcreate(tmp string, path string) (bool, error) {
+	isdir := isDir(path)
+	path = filepath.Join(tmp, filepath.FromSlash(path))
+	if isdir {
+		// Line is a directory.
+		if err := os.MkdirAll(path, 0755); err != nil {
+			return false, err
+		}
+	} else {
+		// Line is a file.
+		if err := os.MkdirAll(filepath.Dir(path), 0755); err != nil {
+			return false, err
+		}
+		f, err := os.Create(path)
+		if err != nil {
+			return false, err
+		}
+		if err := nonil(f.Sync(), f.Close()); err != nil {
+			return false, err
+		}
+	}
+	return isdir, nil
+}
+
+// tmptree TODO
+func tmptree(list string) (string, error) {
+	f, err := os.Open(list)
+	if err != nil {
+		return "", err
+	}
+	defer f.Close()
+	tmp, err := ioutil.TempDir("", "tmptree")
+	if err != nil {
+		return "", err
+	}
+	scanner := bufio.NewScanner(f)
+	for scanner.Scan() {
+		if _, err := tmpcreate(tmp, scanner.Text()); err != nil {
+			return "", err
+		}
+	}
+	if err := scanner.Err(); err != nil {
+		return "", err
+	}
+	Sync()
+	return tmp, nil
+}
+
+type eventinfo struct {
+	sys   interface{}
+	path  string
+	event Event
+	isdir bool
+}
+
+func (ei eventinfo) Sys() interface{} { return ei.sys }
+func (ei eventinfo) Path() string     { return ei.path }
+func (ei eventinfo) Event() Event     { return ei.event }
+func (ei eventinfo) IsDir() bool      { return ei.isdir }
+func (ei eventinfo) String() string   { return fmt.Sprintf("%s on %q", ei.event, ei.path) }
+
+// W TODO
+type W struct {
+	// Watcher TODO
+	Watcher Watcher
+
+	// C TODO
+	C <-chan EventInfo
+
+	// Timeout TODO
+	Timeout time.Duration
+
+	t    *testing.T
+	root string
+}
+
+// newWatcherTest TODO
+func newWatcherTest(t *testing.T, tree string) *W {
+	root, err := tmptree(filepath.FromSlash(tree))
+	if err != nil {
+		t.Fatal(err)
+	}
+	w := &W{
+		t:    t,
+		root: root,
+	}
+	if w, ok := w.watcher().(RecursiveWatcher); ok {
+		if err := w.RecursiveWatch(root, Create|Delete|Move|Write); err != nil {
+			t.Fatal(err)
+		}
+	} else {
+		panic("TODO(rjeczalik)")
+	}
+	return w
+}
+
+func (w *W) watcher() Watcher {
+	if w.Watcher == nil {
+		c := make(chan EventInfo, 128)
+		w.Watcher = NewWatcher(c)
+		w.C = c
+	}
+	return w.Watcher
+}
+
+func (w *W) timeout() time.Duration {
+	if w.Timeout != 0 {
+		return w.Timeout
+	}
+	return 2 * time.Second
+}
+
+// Stop TODO
+func (w *W) Stop() {
+	defer os.RemoveAll(w.root)
+	// TODO(rjeczalik): make Close part of Watcher interface
+	err := w.watcher().(interface {
+		Close() error
+	}).Close()
+	if err != nil {
+		w.t.Fatal(err)
+	}
+}
+
+// create TODO
+func create(w *W, path string) (func(), EventInfo) {
+	ei := eventinfo{path: path, event: Create}
+	fn := func() {
+		var err error
+		if ei.isdir, err = tmpcreate(w.root, filepath.FromSlash(path)); err != nil {
+			w.t.Fatal(err)
+		}
+		Sync()
+	}
+	return fn, ei
+}
+
+// remove TODO
+func remove(w *W, path string) (func(), EventInfo) {
+	ei := eventinfo{path: path, event: Delete}
+	fn := func() {
+		if err := os.RemoveAll(filepath.Join(w.root, filepath.FromSlash(path))); err != nil {
+			w.t.Fatal(err)
+		}
+		Sync()
+	}
+	return fn, ei
+}
+
+// rename TODO
+func rename(w *W, oldpath, newpath string) (func(), EventInfo) {
+	ei := eventinfo{path: newpath, event: Move}
+	fn := func() {
+		err := os.Rename(filepath.Join(w.root, filepath.FromSlash(oldpath)),
+			filepath.Join(w.root, filepath.FromSlash(newpath)))
+		if err != nil {
+			w.t.Fatal(err)
+		}
+		Sync()
+	}
+	return fn, ei
+}
+
+// write TODO
+func write(w *W, path string, p []byte) (func(), EventInfo) {
+	ei := eventinfo{path: path, event: Write}
+	fn := func() {
+		perm := os.FileMode(0644)
+		if isDir(path) {
+			perm = 0755
+		}
+		err := ioutil.WriteFile(filepath.Join(w.root, filepath.FromSlash(path)), p, perm)
+		if err != nil {
+			w.t.Fatal(err)
+		}
+		Sync()
+	}
+	return fn, ei
+}
+
+// Expect TODO
+func (w *W) Expect(fn func(), expected EventInfo) {
+	fn()
+	select {
+	case ei := <-w.C:
+		if ei.Event() != expected.Event() {
+			w.t.Fatalf("want event=%v; got %v", expected.Event(), ei.Event())
+		}
+		if !strings.HasSuffix(ei.Path(), expected.Path()) {
+			w.t.Fatalf("want path=%q; got %q", expected.Path(), ei.Path())
+		}
+	case <-time.After(w.timeout()):
+		w.t.Fatalf("timed out after %v waiting for %v", w.timeout(), expected)
+	}
+}

watcher.go

@@ -6,8 +6,8 @@ package notify
 // filesystem event notification functionalities.
 //
 // The newWatcher function must be implemented by each supported platform.
-func NewWatcher() Watcher {
-	return newWatcher()
+func NewWatcher(c chan<- EventInfo) Watcher {
+	return newWatcher(c)
 }
 
 // Watcher is a temporary interface for wrapping inotify, ReadDirChangesW,

watcher_fsevents.go

@@ -22,6 +22,8 @@ var errDepth = errors.New("exceeded allowed iteration count (circular symlink?)"
 // canonical resolves any symlink in the given path and returns it in a clean form.
 // It expects the path to be absolute. It fails to resolve circular symlinks by
 // maintaining a simple iteration limit.
+//
+// TODO(rjeczalik): handle relative symlinks?
 func canonical(p string) (string, error) {
 	for i, depth := 1, 1; i < len(p); i, depth = i+1, depth+1 {
 		if depth > 128 {
@@ -96,9 +98,10 @@ type fsevents struct {
 	c       chan<- EventInfo
 }
 
-func newWatcher() Watcher {
+func newWatcher(c chan<- EventInfo) Watcher {
 	return &fsevents{
 		watches: make(map[string]*watch),
+		c:       c,
 	}
 }
 
@@ -168,7 +171,7 @@ func (fse *fsevents) Dispatch(c chan<- EventInfo, stop <-chan struct{}) {
 	fse.c = c
 	go func() {
 		<-stop
-		fse.Stop()
+		fse.Close()
 	}()
 }
 
@@ -229,10 +232,11 @@ func (fse *fsevents) RecursiveRewatch(oldpath, newpath string, oldevent, neweven
 	}
 }
 
-// Stop unwatches all watch-points.
-func (fse *fsevents) Stop() {
+// Close unwatches all watch-points.
+func (fse *fsevents) Close() error {
 	for _, w := range fse.watches {
 		w.stream.Stop()
 	}
 	fse.watches = make(map[string]*watch)
+	return nil
 }

watcher_fsevents_test.go

@@ -31,7 +31,7 @@ func symlink(s string) (string, error) {
 	return name, nil
 }
 
-func remove(s ...string) {
+func removeall(s ...string) {
 	for _, s := range s {
 		os.Remove(s)
 	}
@@ -76,7 +76,7 @@ func TestCanonicalizeMultiple(t *testing.T) {
 	if err != nil {
 		t.Fatal(nonil(err, os.Remove(link1)))
 	}
-	defer remove(link1, link2)
+	defer removeall(link1, link2)
 	cases := [...]caseCanonical{
 		{link1, "/private/etc"},
 		{link1 + "/hosts", "/private/etc/hosts"},
@@ -94,7 +94,7 @@ func TestCanonicalizeCircular(t *testing.T) {
 	if err != nil {
 		t.Fatal(nonil(err, os.Remove(tmp1)))
 	}
-	defer remove(tmp1, tmp2)
+	defer removeall(tmp1, tmp2)
 	// Symlink tmp1 -> tmp2.
 	if err = nonil(os.Remove(tmp1), os.Symlink(tmp2, tmp1)); err != nil {
 		t.Fatal(err)

watcher_fsnotify.go

@@ -3,7 +3,6 @@
 package notify
 
 import (
-	"os"
 	"runtime"
 
 	fsnotifyv1 "gopkg.in/fsnotify.v1"
@@ -11,7 +10,8 @@ import (
 
 // Fsnotify implements notify.Watcher interface by wrapping fsnotifyv1 package.
 type fsnotify struct {
-	w *fsnotifyv1.Watcher
+	w    *fsnotifyv1.Watcher
+	stop chan struct{}
 }
 
 // NewWatcher creates new non-recursive watcher backed by fsnotifyv1 package.
@@ -20,8 +20,9 @@ func newWatcher() Watcher {
 	if err != nil {
 		panic(err)
 	}
-	fs := &fsnotify{w: w}
-	runtime.SetFinalizer(fs, (*fsnotify).stop)
+	fs := &fsnotify{w: w, stop: make(chan struct{})}
+	go fs.loop()
+	runtime.SetFinalizer(fs, func(fs fsnotify) { fs.Close() })
 	return fs
 }
 
@@ -36,23 +37,27 @@ func (fs fsnotify) Unwatch(p string) error {
 }
 
 // Dispatch implements notify.Watcher interface.
+//
+// TODO(rjeczalik): remove
 func (fs fsnotify) Dispatch(c chan<- EventInfo, stop <-chan struct{}) {
-	go func() {
-		for {
-			select {
-			case e := <-fs.w.Events:
-				c <- newEvent(e)
-			case <-stop:
-				fs.stop()
-				return
-			}
-		}
-	}()
 }
 
-func (fs *fsnotify) stop() {
+func (fs fsnotify) loop() {
+	for {
+		select {
+		case e := <-fs.w.Events:
+			c <- newEvent(e)
+		case <-w.stop:
+			return
+		}
+	}
+}
+
+func (fs *fsnotify) Close() error {
 	if fs.w != nil {
 		fs.w.Close()
 		fs.w = nil
+		close(fs.stop)
 	}
+	return nil
 }

watcher_kqueue.go

@@ -15,26 +15,38 @@ import (
 // TODO: Take into account currently monitored files with those read from dir.
 
 // newWatcher returns `kqueue` Watcher implementation.
-func newWatcher() Watcher {
+func newWatcher(c chan<- EventInfo) Watcher {
 	k := &kqueue{
 		idLkp:  make(map[int]*watched, 0),
 		pthLkp: make(map[string]*watched, 0),
+		c:      c,
+		s:      make(chan struct{}),
 	}
 	if err := k.init(); err != nil {
 		// TODO: Does it really has to be this way?
 		panic(err)
 	}
+	go k.monitor()
 	return k
 }
 
-// stop closes all still open file descriptors and kqueue.
-func (k *kqueue) stop() {
-	for i := range k.idLkp {
-		syscall.Close(k.idLkp[i].fd)
+// Close closes all still open file descriptors and kqueue.
+func (k *kqueue) Close() error {
+	k.Lock()
+	if k.s != nil {
+		close(k.s)
+		k.s = nil
+	}
+	for _, w := range k.idLkp {
+		syscall.Close(w.fd)
 	}
 	if k.fd != nil {
 		syscall.Close(*k.fd)
+		k.fd = nil
 	}
+	k.idLkp, k.pthLkp = nil, nil
+	k.Unlock()
+	return nil
 }
 
 // sendEvents sends reported events one by one through chan.
@@ -99,7 +111,6 @@ func (k *kqueue) monitor() {
 		var kevn [1]syscall.Kevent_t
 		select {
 		case <-k.s:
-			k.stop()
 			return
 		default:
 		}
@@ -176,7 +187,7 @@ type kqueue struct {
 	// c is a channel used to pass events further.
 	c chan<- EventInfo
 	// s is a channel used to stop monitoring.
-	s <-chan struct{}
+	s chan struct{}
 }
 
 // watched is a data structure representing watched file/directory.
@@ -258,7 +269,7 @@ func (k *kqueue) singlewatch(p string, e Event, direct, dir bool) error {
 }
 
 // unwatch stops watching `p` file/directory.
-func (k *kqueue) singleunwatch(p string, direct bool) (err error) {
+func (k *kqueue) singleunwatch(p string, direct bool) error {
 	w := k.pthLkp[p]
 	if w == nil {
 		return errNotWatched
@@ -270,37 +281,36 @@ func (k *kqueue) singleunwatch(p string, direct bool) (err error) {
 	}
 	var kevn [1]syscall.Kevent_t
 	syscall.SetKevent(&kevn[0], w.fd, syscall.EVFILT_VNODE, syscall.EV_DELETE)
-	if _, err = syscall.Kevent(*k.fd, kevn[:], nil, nil); err != nil {
-		return
+	if _, err := syscall.Kevent(*k.fd, kevn[:], nil, nil); err != nil {
+		return err
 	}
 	if w.eNonDir&w.eDir != 0 {
-		if err = k.singlewatch(p, w.eNonDir|w.eDir, w.eNonDir == 0, w.dir); err != nil {
-			return
+		if err := k.singlewatch(p, w.eNonDir|w.eDir, w.eNonDir == 0, w.dir); err != nil {
+			return err
 		}
 	} else {
 		k.del(w)
 	}
-	return
+	return nil
 }
 
 // walk runs `f` func on each file from `p` directory.
-func (k *kqueue) walk(p string, f func(os.FileInfo) error) (err error) {
-	var fp *os.File
-	if fp, err = os.Open(p); err != nil {
-		return
-	}
-	var ls []os.FileInfo
-	if ls, err = fp.Readdir(-1); err != nil {
-		fp.Close()
-		return
+func (k *kqueue) walk(p string, f func(os.FileInfo) error) error {
+	fp, err := os.Open(p)
+	if err != nil {
+		return err
 	}
+	ls, err := fp.Readdir(0)
 	fp.Close()
+	if err != nil {
+		return err
+	}
 	for i := range ls {
-		if err = f(ls[i]); err != nil {
-			return
+		if err := f(ls[i]); err != nil {
+			return err
 		}
 	}
-	return
+	return nil
 }
 
 func (k *kqueue) unwatch(p string, isdir bool) error {
@@ -371,7 +381,7 @@ func isdir(p string) (bool, error) {
 }
 
 // Dispatch implements `Watcher` interface.
+//
+// TODO(rjeczalik): remove
 func (k *kqueue) Dispatch(c chan<- EventInfo, stop <-chan struct{}) {
-	k.c, k.s = c, stop
-	go k.monitor()
 }

watcher_old_test.go

@@ -1,4 +1,4 @@
-// +build !darwin
+// +build ignore
 
 package notify_test
 

watcher_test.go

@@ -1 +1,14 @@
+// +build darwin
+
 package notify
+
+import "testing"
+
+// TODO(rjeczalik): add more test-cases
+
+func TestWatcher(t *testing.T) {
+	w := newWatcherTest(t, "testdata/gopath.txt")
+	defer w.Stop()
+
+	w.Expect(create(w, "src/github.com/rjeczalik/which/.which.go.swp"))
+}

watcher_windows.go

@@ -242,10 +242,11 @@ type watcher struct {
 }
 
 // NewWatcher creates new non-recursive watcher backed by ReadDirectoryChangesW.
-func newWatcher() *watcher {
+func newWatcher(c chan<- EventInfo) *watcher {
 	return &watcher{
 		m:   make(map[string]*watched),
 		cph: syscall.InvalidHandle,
+		c:   c,
 	}
 }
 
@@ -494,6 +495,8 @@ func (w *watcher) unwatch(path string) (err error) {
 }
 
 // Dispatch implements notify.Watcher interface.
+//
+// TODO(rjeczalik): remove
 func (w *watcher) Dispatch(c chan<- EventInfo, stop <-chan struct{}) {
 	w.c = c
 }