From 6aef373ea6eabb67a4c87ed60852354fc6e02b25 Mon Sep 17 00:00:00 2001 From: miraclesu Date: Wed, 9 Dec 2015 15:29:06 +0800 Subject: [PATCH] fix block until exists function --- tail_test.go | 24 +++++- watch/inotify.go | 11 +-- watch/inotify_tracker.go | 153 ++++++++++++++++++++++++++++++--------- 3 files changed, 146 insertions(+), 42 deletions(-) diff --git a/tail_test.go b/tail_test.go index b2f22a5..b6d0676 100644 --- a/tail_test.go +++ b/tail_test.go @@ -155,7 +155,7 @@ func TestLocationEnd(_t *testing.T) { func TestLocationMiddle(_t *testing.T) { // Test reading from middle. - t := NewTailTest("location-end", _t) + t := NewTailTest("location-middle", _t) t.CreateFile("test.txt", "hello\nworld\n") tail := t.StartTail("test.txt", Config{Follow: true, Location: &SeekInfo{-6, os.SEEK_END}}) go t.VerifyTailOutput(tail, []string{"world", "more", "data"}) @@ -331,6 +331,28 @@ func TestTell(_t *testing.T) { tail.Cleanup() } +func TestBlockUntilExists(_t *testing.T) { + t := NewTailTest("block-until-file-exists", _t) + config := Config{ + Follow: true, + } + tail := t.StartTail("test.txt", config) + go func() { + time.Sleep(100 * time.Millisecond) + t.CreateFile("test.txt", "hello world\n") + }() + for l := range tail.Lines { + if l.Text != "hello world" { + t.Fatalf("mismatch; expected hello world, but got %s", + l.Text) + } + break + } + t.RemoveFile("test.txt") + tail.Stop() + tail.Cleanup() +} + // Test library type TailTest struct { diff --git a/watch/inotify.go b/watch/inotify.go index d6635f4..8d11f10 100644 --- a/watch/inotify.go +++ b/watch/inotify.go @@ -20,19 +20,16 @@ type InotifyFileWatcher struct { } func NewInotifyFileWatcher(filename string) *InotifyFileWatcher { - fw := &InotifyFileWatcher{filename, 0} + fw := &InotifyFileWatcher{filepath.Clean(filename), 0} return fw } func (fw *InotifyFileWatcher) BlockUntilExists(t *tomb.Tomb) error { - dirname := filepath.Dir(fw.Filename) - - // Watch for new files to be created in the parent directory. - err := Watch(dirname) + err := WatchCreate(fw.Filename) if err != nil { return err } - defer RemoveWatch(dirname) + defer RemoveWatchCreate(fw.Filename) // Do a real check now as the file might have been created before // calling `WatchFlags` above. @@ -48,7 +45,7 @@ func (fw *InotifyFileWatcher) BlockUntilExists(t *tomb.Tomb) error { case evt, ok := <-events: if !ok { return fmt.Errorf("inotify watcher has been closed") - } else if evt.Name == fw.Filename { + } else if filepath.Clean(evt.Name) == fw.Filename { return nil } case <-t.Dying(): diff --git a/watch/inotify_tracker.go b/watch/inotify_tracker.go index 07e7d7a..f53f2c3 100644 --- a/watch/inotify_tracker.go +++ b/watch/inotify_tracker.go @@ -5,6 +5,7 @@ package watch import ( "log" "os" + "path/filepath" "sync" "syscall" @@ -14,19 +15,25 @@ import ( ) type InotifyTracker struct { - mux sync.Mutex - watcher *fsnotify.Watcher - chans map[string]chan fsnotify.Event - done map[string]chan bool - watch chan *watchInfo - remove chan string - error chan error + mux sync.Mutex + watcher *fsnotify.Watcher + chans map[string]chan fsnotify.Event + done map[string]chan bool + watchNums map[string]int + watch chan *watchInfo + remove chan *watchInfo + error chan error } type watchInfo struct { + op fsnotify.Op fname string } +func (this *watchInfo) isCreate() bool { + return this.op == fsnotify.Create +} + var ( // globally shared InotifyTracker; ensures only one fsnotify.Watcher is used shared *InotifyTracker @@ -35,12 +42,13 @@ var ( once = sync.Once{} goRun = func() { shared = &InotifyTracker{ - mux: sync.Mutex{}, - chans: make(map[string]chan fsnotify.Event), - done: make(map[string]chan bool), - watch: make(chan *watchInfo), - remove: make(chan string), - error: make(chan error), + mux: sync.Mutex{}, + chans: make(map[string]chan fsnotify.Event), + done: make(map[string]chan bool), + watchNums: make(map[string]int), + watch: make(chan *watchInfo), + remove: make(chan *watchInfo), + error: make(chan error), } go shared.run() } @@ -50,29 +58,58 @@ var ( // Watch signals the run goroutine to begin watching the input filename func Watch(fname string) error { + return watch(&watchInfo{ + fname: fname, + }) +} + +// Watch create signals the run goroutine to begin watching the input filename +// if call the WatchCreate function, don't call the Cleanup, call the RemoveWatchCreate +func WatchCreate(fname string) error { + return watch(&watchInfo{ + op: fsnotify.Create, + fname: fname, + }) +} + +func watch(winfo *watchInfo) error { // start running the shared InotifyTracker if not already running once.Do(goRun) - shared.watch <- &watchInfo{ - fname: fname, - } + winfo.fname = filepath.Clean(winfo.fname) + shared.watch <- winfo return <-shared.error } // RemoveWatch signals the run goroutine to remove the watch for the input filename func RemoveWatch(fname string) { + remove(&watchInfo{ + fname: fname, + }) +} + +// RemoveWatch create signals the run goroutine to remove the watch for the input filename +func RemoveWatchCreate(fname string) { + remove(&watchInfo{ + op: fsnotify.Create, + fname: fname, + }) +} + +func remove(winfo *watchInfo) { // start running the shared InotifyTracker if not already running once.Do(goRun) + winfo.fname = filepath.Clean(winfo.fname) shared.mux.Lock() - done := shared.done[fname] + done := shared.done[winfo.fname] if done != nil { - delete(shared.done, fname) + delete(shared.done, winfo.fname) close(done) } shared.mux.Unlock() - shared.remove <- fname + shared.remove <- winfo } // Events returns a channel to which FileEvents corresponding to the input filename @@ -92,36 +129,84 @@ func Cleanup(fname string) { // watchFlags calls fsnotify.WatchFlags for the input filename and flags, creating // a new Watcher if the previous Watcher was closed. -func (shared *InotifyTracker) addWatch(fname string) error { +func (shared *InotifyTracker) addWatch(winfo *watchInfo) error { shared.mux.Lock() defer shared.mux.Unlock() - if shared.chans[fname] == nil { - shared.chans[fname] = make(chan fsnotify.Event) - shared.done[fname] = make(chan bool) + if shared.chans[winfo.fname] == nil { + shared.chans[winfo.fname] = make(chan fsnotify.Event) + shared.done[winfo.fname] = make(chan bool) } - return shared.watcher.Add(fname) + + fname := winfo.fname + if winfo.isCreate() { + // Watch for new files to be created in the parent directory. + fname = filepath.Dir(fname) + } + + // already in inotify watch + if shared.watchNums[fname] > 0 { + shared.watchNums[fname]++ + if winfo.isCreate() { + shared.watchNums[winfo.fname]++ + } + return nil + } + + err := shared.watcher.Add(fname) + if err == nil { + shared.watchNums[fname]++ + if winfo.isCreate() { + shared.watchNums[winfo.fname]++ + } + } + return err } // removeWatch calls fsnotify.RemoveWatch for the input filename and closes the // corresponding events channel. -func (shared *InotifyTracker) removeWatch(fname string) { +func (shared *InotifyTracker) removeWatch(winfo *watchInfo) { shared.mux.Lock() defer shared.mux.Unlock() - if ch := shared.chans[fname]; ch != nil { - shared.watcher.Remove(fname) + ch := shared.chans[winfo.fname] + if ch == nil { + return + } - delete(shared.chans, fname) - close(ch) + fname := winfo.fname + if winfo.isCreate() { + // Watch for new files to be created in the parent directory. + fname = filepath.Dir(fname) + } + + shared.watchNums[fname]-- + if shared.watchNums[fname] == 0 { + delete(shared.watchNums, fname) + // TODO: handle error + shared.watcher.Remove(fname) + } + + delete(shared.chans, winfo.fname) + close(ch) + + if !winfo.isCreate() { + return + } + + shared.watchNums[winfo.fname]-- + if shared.watchNums[winfo.fname] == 0 { + delete(shared.watchNums, winfo.fname) } } // sendEvent sends the input event to the appropriate Tail. func (shared *InotifyTracker) sendEvent(event fsnotify.Event) { + name := filepath.Clean(event.Name) + shared.mux.Lock() - ch := shared.chans[event.Name] - done := shared.done[event.Name] + ch := shared.chans[name] + done := shared.done[name] shared.mux.Unlock() if ch != nil && done != nil { @@ -144,10 +229,10 @@ func (shared *InotifyTracker) run() { for { select { case winfo := <-shared.watch: - shared.error <- shared.addWatch(winfo.fname) + shared.error <- shared.addWatch(winfo) - case fname := <-shared.remove: - shared.removeWatch(fname) + case winfo := <-shared.remove: + shared.removeWatch(winfo) case event, open := <-shared.watcher.Events: if !open {