Merge pull request #72 from miraclesu/fix/watch_create

fix block until exists function

.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
+  - go test -race -v ./...
-  - GOPATH=$HOME/gopath $HOME/gopath/bin/depman
-  - go test -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

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.colseFile()
-		tail.file.Close()
-	}
 	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 {

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 {

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)
+	err := WatchCreate(fw.Filename)
-
-	// Watch for new files to be created in the parent directory.
-	err := Watch(dirname)
 	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 {
+func (fw *InotifyFileWatcher) ChangeEvents(t *tomb.Tomb, pos int64) (*FileChanges, error) {
-	changes := NewFileChanges()
-
 	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
 }

watch/inotify_tracker.go

@@ -5,6 +5,7 @@ package watch
 import (
 	"log"
 	"os"
+	"path/filepath"
 	"sync"
 	"syscall"
 
@@ -18,15 +19,21 @@ type InotifyTracker struct {
 	watcher   *fsnotify.Watcher
 	chans     map[string]chan fsnotify.Event
 	done      map[string]chan bool
+	watchNums map[string]int
 	watch     chan *watchInfo
-	remove  chan string
+	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
@@ -38,8 +45,9 @@ var (
 			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 string),
+			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{
+	winfo.fname = filepath.Clean(winfo.fname)
-		fname: 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 {
+	if shared.chans[winfo.fname] == nil {
-		shared.chans[fname] = make(chan fsnotify.Event)
+		shared.chans[winfo.fname] = make(chan fsnotify.Event)
-		shared.done[fname] = make(chan bool)
+		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 {
+	ch := shared.chans[winfo.fname]
-		shared.watcher.Remove(fname)
+	if ch == nil {
+		return
+	}
 
-		delete(shared.chans, fname)
+	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]
+	ch := shared.chans[name]
-	done := shared.done[event.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:
+		case winfo := <-shared.remove:
-			shared.removeWatch(fname)
+			shared.removeWatch(winfo)
 
 		case event, open := <-shared.watcher.Events:
 			if !open {

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() {

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)
 }