go-kit
/
tail
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge pull request #53 from wagerlabs/v1 

Single inotify instance with multiple watchers
This commit is contained in:
Sridhar Ratnakumar 2015-07-03 14:23:58 -07:00
parent 15e71d383e 244ac5d165
commit b1f6c8f553
3 changed files with 45 additions and 51 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
tail.go
View File

@ -5,16 +5,17 @@ package tail
import (
"bufio"
"fmt"
"github.com/ActiveState/tail/ratelimiter"
"github.com/ActiveState/tail/util"
"github.com/ActiveState/tail/watch"
"gopkg.in/tomb.v1"
"io"
"io/ioutil"
"log"
"os"
"strings"
"time"
"github.com/ActiveState/tail/ratelimiter"
"github.com/ActiveState/tail/util"
"github.com/ActiveState/tail/watch"
"gopkg.in/tomb.v1"
)
var (
@ -63,6 +64,8 @@ type Tail struct {
file *os.File
reader *bufio.Reader
tracker *watch.InotifyTracker
watcher watch.FileWatcher
changes *watch.FileChanges
@ -99,7 +102,12 @@ func TailFile(filename string, config Config) (*Tail, error) {
if t.Poll {
t.watcher = watch.NewPollingFileWatcher(filename)
} else {
t.watcher = watch.NewInotifyFileWatcher(filename)
t.tracker = watch.NewInotifyTracker()
w, err := t.tracker.NewWatcher()
if err != nil {
return nil, err
}
t.watcher = watch.NewInotifyFileWatcher(filename, w)
}
if t.MustExist {
@ -381,6 +389,8 @@ func (tail *Tail) sendLine(line string) bool {
// Cleanup removes inotify watches added by the tail package. This function is
// meant to be invoked from a process's exit handler. Linux kernel may not
// automatically remove inotify watches after the process exits.
func Cleanup() {
watch.Cleanup()
func (tail *Tail) Cleanup() {
if tail.tracker != nil {
tail.tracker.CloseAll()
}
}

31
tail_test.go
View File

@ -6,14 +6,15 @@
package tail
import (
"./watch"
_ "fmt"
"github.com/ActiveState/tail/ratelimiter"
"io/ioutil"
"os"
"strings"
"testing"
"time"
"./watch"
"github.com/ActiveState/tail/ratelimiter"
)
func init() {
@ -40,7 +41,7 @@ func TestMustExist(t *testing.T) {
t.Error("MustExist:true on an existing file is violated")
}
tail.Stop()
Cleanup()
tail.Cleanup()
}
func TestStop(t *testing.T) {
@ -51,7 +52,7 @@ func TestStop(t *testing.T) {
if tail.Stop() != nil {
t.Error("Should be stoped successfully")
}
Cleanup()
tail.Cleanup()
}
func MaxLineSizeT(_t *testing.T, follow bool, fileContent string, expected []string) {
@ -65,7 +66,7 @@ func MaxLineSizeT(_t *testing.T, follow bool, fileContent string, expected []str
<-time.After(100 * time.Millisecond)
t.RemoveFile("test.txt")
tail.Stop()
Cleanup()
tail.Cleanup()
}
func TestMaxLineSizeFollow(_t *testing.T) {
@ -89,7 +90,7 @@ func TestOver4096ByteLine(_t *testing.T) {
<-time.After(100 * time.Millisecond)
t.RemoveFile("test.txt")
tail.Stop()
Cleanup()
tail.Cleanup()
}
func TestOver4096ByteLineWithSetMaxLineSize(_t *testing.T) {
t := NewTailTest("Over4096ByteLineMaxLineSize", _t)
@ -103,7 +104,7 @@ func TestOver4096ByteLineWithSetMaxLineSize(_t *testing.T) {
<-time.After(100 * time.Millisecond)
t.RemoveFile("test.txt")
// tail.Stop()
Cleanup()
tail.Cleanup()
}
func TestLocationFull(_t *testing.T) {
@ -117,7 +118,7 @@ func TestLocationFull(_t *testing.T) {
<-time.After(100 * time.Millisecond)
t.RemoveFile("test.txt")
tail.Stop()
Cleanup()
tail.Cleanup()
}
func TestLocationFullDontFollow(_t *testing.T) {
@ -132,7 +133,7 @@ func TestLocationFullDontFollow(_t *testing.T) {
<-time.After(100 * time.Millisecond)
tail.Stop()
Cleanup()
tail.Cleanup()
}
func TestLocationEnd(_t *testing.T) {
@ -149,7 +150,7 @@ func TestLocationEnd(_t *testing.T) {
<-time.After(100 * time.Millisecond)
t.RemoveFile("test.txt")
tail.Stop()
Cleanup()
tail.Cleanup()
}
func TestLocationMiddle(_t *testing.T) {
@ -167,7 +168,7 @@ func TestLocationMiddle(_t *testing.T) {
<-time.After(100 * time.Millisecond)
t.RemoveFile("test.txt")
tail.Stop()
Cleanup()
tail.Cleanup()
}
func _TestReOpen(_t *testing.T, poll bool) {
@ -210,7 +211,7 @@ func _TestReOpen(_t *testing.T, poll bool) {
// the reading of data written above. Timings can vary based on
// test environment.
// tail.Stop()
Cleanup()
tail.Cleanup()
}
// The use of polling file watcher could affect file rotation
@ -253,7 +254,7 @@ func _TestReSeek(_t *testing.T, poll bool) {
// the reading of data written above. Timings can vary based on
// test environment.
// tail.Stop()
Cleanup()
tail.Cleanup()
}
// The use of polling file watcher could affect file rotation
@ -293,7 +294,7 @@ func TestRateLimiting(_t *testing.T) {
t.RemoveFile("test.txt")
// tail.Stop()
Cleanup()
tail.Cleanup()
}
func TestTell(_t *testing.T) {
@ -327,7 +328,7 @@ func TestTell(_t *testing.T) {
}
t.RemoveFile("test.txt")
tail.Done()
Cleanup()
tail.Cleanup()
}
// Test library

41
watch/inotify.go
View File

@ -4,40 +4,35 @@ package watch
import (
"fmt"
"os"
"path/filepath"
"github.com/ActiveState/tail/util"
"gopkg.in/fsnotify.v0"
"gopkg.in/tomb.v1"
"os"
"path/filepath"
)
var inotifyTracker *InotifyTracker
// InotifyFileWatcher uses inotify to monitor file changes.
type InotifyFileWatcher struct {
Filename string
Size int64
w *fsnotify.Watcher
}
func NewInotifyFileWatcher(filename string) *InotifyFileWatcher {
fw := &InotifyFileWatcher{filename, 0}
func NewInotifyFileWatcher(filename string, w *fsnotify.Watcher) *InotifyFileWatcher {
fw := &InotifyFileWatcher{filename, 0, w}
return fw
}
func (fw *InotifyFileWatcher) BlockUntilExists(t *tomb.Tomb) error {
w, err := inotifyTracker.NewWatcher()
if err != nil {
return err
}
defer inotifyTracker.CloseWatcher(w)
dirname := filepath.Dir(fw.Filename)
// Watch for new files to be created in the parent directory.
err = w.WatchFlags(dirname, fsnotify.FSN_CREATE)
err := fw.w.WatchFlags(dirname, fsnotify.FSN_CREATE)
if err != nil {
return err
}
defer fw.w.RemoveWatch(dirname)
// Do a real check now as the file might have been created before
// calling `WatchFlags` above.
@ -48,7 +43,7 @@ func (fw *InotifyFileWatcher) BlockUntilExists(t *tomb.Tomb) error {
for {
select {
case evt, ok := <-w.Event:
case evt, ok := <-fw.w.Event:
if !ok {
return fmt.Errorf("inotify watcher has been closed")
} else if evt.Name == fw.Filename {
@ -64,11 +59,7 @@ func (fw *InotifyFileWatcher) BlockUntilExists(t *tomb.Tomb) error {
func (fw *InotifyFileWatcher) ChangeEvents(t *tomb.Tomb, fi os.FileInfo) *FileChanges {
changes := NewFileChanges()
w, err := inotifyTracker.NewWatcher()
if err != nil {
util.Fatal("Error creating fsnotify watcher: %v", err)
}
err = w.Watch(fw.Filename)
err := fw.w.Watch(fw.Filename)
if err != nil {
util.Fatal("Error watching %v: %v", fw.Filename, err)
}
@ -76,7 +67,7 @@ func (fw *InotifyFileWatcher) ChangeEvents(t *tomb.Tomb, fi os.FileInfo) *FileCh
fw.Size = fi.Size()
go func() {
defer inotifyTracker.CloseWatcher(w)
defer fw.w.RemoveWatch(fw.Filename)
defer changes.Close()
for {
@ -86,7 +77,7 @@ func (fw *InotifyFileWatcher) ChangeEvents(t *tomb.Tomb, fi os.FileInfo) *FileCh
var ok bool
select {
case evt, ok = <-w.Event:
case evt, ok = <-fw.w.Event:
if !ok {
return
}
@ -126,11 +117,3 @@ func (fw *InotifyFileWatcher) ChangeEvents(t *tomb.Tomb, fi os.FileInfo) *FileCh
return changes
}
func Cleanup() {
inotifyTracker.CloseAll()
}
func init() {
inotifyTracker = NewInotifyTracker()
}