diff --git a/.travis.yml b/.travis.yml index 57c00e2..15fbdbc 100644 --- a/.travis.yml +++ b/.travis.yml @@ -1,14 +1,13 @@ language: go -# depman needs to be installed with GOPATH pointing to a single directory. script: - - GOPATH=$HOME/gopath go get github.com/vube/depman - - GOPATH=$HOME/gopath $HOME/gopath/bin/depman - - go test -v ./... + - go test -race -v ./... go: - 1.3.3 - - 1.4.2 + - 1.4.3 + - 1.5.2 install: - go get gopkg.in/fsnotify.v1 + - go get gopkg.in/tomb.v1 diff --git a/tail.go b/tail.go index ae5b95b..1bae11c 100644 --- a/tail.go +++ b/tail.go @@ -10,6 +10,7 @@ import ( "log" "os" "strings" + "sync" "time" "github.com/hpcloud/tail/ratelimiter" @@ -82,6 +83,8 @@ type Tail struct { changes *watch.FileChanges tomb.Tomb // provides: Done, Kill, Dying + + lk sync.Mutex } var ( @@ -140,7 +143,9 @@ func (tail *Tail) Tell() (offset int64, err error) { } offset, err = tail.file.Seek(0, os.SEEK_CUR) if err == nil { + tail.lk.Lock() offset -= int64(tail.reader.Buffered()) + tail.lk.Unlock() } return } @@ -153,15 +158,18 @@ func (tail *Tail) Stop() error { func (tail *Tail) close() { close(tail.Lines) + tail.colseFile() +} + +func (tail *Tail) colseFile() { if tail.file != nil { tail.file.Close() + tail.file = nil } } func (tail *Tail) reopen() error { - if tail.file != nil { - tail.file.Close() - } + tail.colseFile() for { var err error tail.file, err = OpenFile(tail.Filename) @@ -184,7 +192,9 @@ func (tail *Tail) reopen() error { } func (tail *Tail) readLine() (string, error) { + tail.lk.Lock() line, err := tail.reader.ReadString('\n') + tail.lk.Unlock() if err != nil { // Note ReadString "returns the data read before the error" in // case of an error, including EOF, so we return it as is. The @@ -312,7 +322,10 @@ func (tail *Tail) waitForChanges() error { if err != nil { return err } - tail.changes = tail.watcher.ChangeEvents(&tail.Tomb, pos) + tail.changes, err = tail.watcher.ChangeEvents(&tail.Tomb, pos) + if err != nil { + return err + } } select { 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..58efc10 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(): @@ -58,14 +55,13 @@ func (fw *InotifyFileWatcher) BlockUntilExists(t *tomb.Tomb) error { panic("unreachable") } -func (fw *InotifyFileWatcher) ChangeEvents(t *tomb.Tomb, pos int64) *FileChanges { - changes := NewFileChanges() - +func (fw *InotifyFileWatcher) ChangeEvents(t *tomb.Tomb, pos int64) (*FileChanges, error) { err := Watch(fw.Filename) if err != nil { - go changes.NotifyDeleted() + return nil, err } + changes := NewFileChanges() fw.Size = pos go func() { @@ -119,5 +115,5 @@ func (fw *InotifyFileWatcher) ChangeEvents(t *tomb.Tomb, pos int64) *FileChanges } }() - return changes + return changes, nil } 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 { diff --git a/watch/polling.go b/watch/polling.go index 5479272..33b8b4d 100644 --- a/watch/polling.go +++ b/watch/polling.go @@ -3,11 +3,12 @@ package watch import ( - "github.com/hpcloud/tail/util" - "gopkg.in/tomb.v1" "os" "runtime" "time" + + "github.com/hpcloud/tail/util" + "gopkg.in/tomb.v1" ) // PollingFileWatcher polls the file for changes. @@ -40,7 +41,12 @@ func (fw *PollingFileWatcher) BlockUntilExists(t *tomb.Tomb) error { panic("unreachable") } -func (fw *PollingFileWatcher) ChangeEvents(t *tomb.Tomb, pos int64) *FileChanges { +func (fw *PollingFileWatcher) ChangeEvents(t *tomb.Tomb, pos int64) (*FileChanges, error) { + origFi, err := os.Stat(fw.Filename) + if err != nil { + return nil, err + } + changes := NewFileChanges() var prevModTime time.Time @@ -48,11 +54,6 @@ func (fw *PollingFileWatcher) ChangeEvents(t *tomb.Tomb, pos int64) *FileChanges // the fatal (below) with tomb's Kill. fw.Size = pos - origFi, err := os.Stat(fw.Filename) - if err != nil { - changes.NotifyDeleted() - return changes - } go func() { defer changes.Close() @@ -110,7 +111,7 @@ func (fw *PollingFileWatcher) ChangeEvents(t *tomb.Tomb, pos int64) *FileChanges } }() - return changes + return changes, nil } func init() { diff --git a/watch/watch.go b/watch/watch.go index fbc08a5..05baea1 100644 --- a/watch/watch.go +++ b/watch/watch.go @@ -15,5 +15,5 @@ type FileWatcher interface { // or truncation event. // In order to properly report truncations, ChangeEvents requires // the caller to pass their current offset in the file. - ChangeEvents(*tomb.Tomb, int64) *FileChanges + ChangeEvents(*tomb.Tomb, int64) (*FileChanges, error) }