diff --git a/notify.go b/notify.go index d154bb7..686ca27 100644 --- a/notify.go +++ b/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 diff --git a/sync_unix_test.go b/sync_unix_test.go new file mode 100644 index 0000000..e14da4e --- /dev/null +++ b/sync_unix_test.go @@ -0,0 +1,10 @@ +// +build !windows + +package notify + +import "golang.org/x/sys/unix" + +// TODO +func Sync() { + unix.Sync() +} diff --git a/sync_windows_test.go b/sync_windows_test.go new file mode 100644 index 0000000..55237ae --- /dev/null +++ b/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) + } +} diff --git a/test_test.go b/test_test.go new file mode 100644 index 0000000..7185c9f --- /dev/null +++ b/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) + } +} diff --git a/watcher.go b/watcher.go index bdf4866..2051922 100644 --- a/watcher.go +++ b/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, diff --git a/watcher_fsevents.go b/watcher_fsevents.go index b1ebc5b..a0d2873 100644 --- a/watcher_fsevents.go +++ b/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 } diff --git a/watcher_fsevents_test.go b/watcher_fsevents_test.go index 464b389..e6f52b0 100644 --- a/watcher_fsevents_test.go +++ b/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) diff --git a/watcher_fsnotify.go b/watcher_fsnotify.go index da14a1c..ec95d1c 100644 --- a/watcher_fsnotify.go +++ b/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 } diff --git a/watcher_kqueue.go b/watcher_kqueue.go index 0406387..e1a0491 100644 --- a/watcher_kqueue.go +++ b/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() } diff --git a/watcher_old_test.go b/watcher_old_test.go index c20de80..6aa1603 100644 --- a/watcher_old_test.go +++ b/watcher_old_test.go @@ -1,4 +1,4 @@ -// +build !darwin +// +build ignore package notify_test diff --git a/watcher_test.go b/watcher_test.go index a3131f1..96a7dd3 100644 --- a/watcher_test.go +++ b/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")) +} diff --git a/watcher_windows.go b/watcher_windows.go index 3758987..8d8e0b4 100644 --- a/watcher_windows.go +++ b/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 }