fix block until exists function

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

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
+	mux       sync.Mutex
-	watcher *fsnotify.Watcher
+	watcher   *fsnotify.Watcher
-	chans   map[string]chan fsnotify.Event
+	chans     map[string]chan fsnotify.Event
-	done    map[string]chan bool
+	done      map[string]chan bool
-	watch   chan *watchInfo
+	watchNums map[string]int
-	remove  chan string
+	watch     chan *watchInfo
-	error   chan error
+	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{},
+			mux:       sync.Mutex{},
-			chans:  make(map[string]chan fsnotify.Event),
+			chans:     make(map[string]chan fsnotify.Event),
-			done:   make(map[string]chan bool),
+			done:      make(map[string]chan bool),
-			watch:  make(chan *watchInfo),
+			watchNums: make(map[string]int),
-			remove: make(chan string),
+			watch:     make(chan *watchInfo),
-			error:  make(chan error),
+			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
-		close(ch)
+	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 {