make watch.go its own package

for the purpose for upcoming changes to the watch functionality.
This commit is contained in:
Sridhar Ratnakumar 2013-05-28 16:34:36 -07:00
parent b664e9fc9d
commit 0f67bc352f
6 changed files with 240 additions and 223 deletions

View File

@ -10,6 +10,7 @@ import (
"log" "log"
"os" "os"
"time" "time"
"github.com/ActiveState/tail/watch"
) )
type Line struct { type Line struct {
@ -34,7 +35,7 @@ type Tail struct {
file *os.File file *os.File
reader *bufio.Reader reader *bufio.Reader
watcher FileWatcher watcher watch.FileWatcher
tomb.Tomb // provides: Done, Kill, Dying tomb.Tomb // provides: Done, Kill, Dying
} }
@ -62,9 +63,9 @@ func TailFile(filename string, config Config) (*Tail, error) {
Config: config} Config: config}
if t.Poll { if t.Poll {
t.watcher = NewPollingFileWatcher(filename) t.watcher = watch.NewPollingFileWatcher(filename)
} else { } else {
t.watcher = NewInotifyFileWatcher(filename) t.watcher = watch.NewInotifyFileWatcher(filename)
} }
if t.MustExist { if t.MustExist {

View File

@ -11,6 +11,7 @@ import (
"os" "os"
"testing" "testing"
"time" "time"
"./watch"
) )
func init() { func init() {
@ -104,7 +105,7 @@ func _TestReOpen(_t *testing.T, poll bool) {
if poll { if poll {
// Give polling a chance to read the just-written lines (more; // Give polling a chance to read the just-written lines (more;
// data), before we recreate the file again below. // data), before we recreate the file again below.
<-time.After(POLL_DURATION) <-time.After(watch.POLL_DURATION)
} }
// rename must trigger reopen // rename must trigger reopen
@ -114,7 +115,7 @@ func _TestReOpen(_t *testing.T, poll bool) {
if poll { if poll {
// This time, wait a bit before creating the file to test // This time, wait a bit before creating the file to test
// PollingFileWatcher's BlockUntilExists. // PollingFileWatcher's BlockUntilExists.
<-time.After(POLL_DURATION) <-time.After(watch.POLL_DURATION)
} }
t.CreateFile("test.txt", "endofworld") t.CreateFile("test.txt", "endofworld")
@ -159,7 +160,7 @@ func _TestReSeek(_t *testing.T, poll bool) {
if poll { if poll {
// Give polling a chance to read the just-written lines (more; // Give polling a chance to read the just-written lines (more;
// data), before we truncate the file again below. // data), before we truncate the file again below.
<-time.After(POLL_DURATION) <-time.After(watch.POLL_DURATION)
} }
println("truncating..") println("truncating..")
t.TruncateFile("test.txt", "h311o\nw0r1d\nendofworld\n") t.TruncateFile("test.txt", "h311o\nw0r1d\nendofworld\n")
@ -167,7 +168,7 @@ func _TestReSeek(_t *testing.T, poll bool) {
if poll { if poll {
// Give polling a chance to read the just-written lines (more; // Give polling a chance to read the just-written lines (more;
// data), before we recreate the file again below. // data), before we recreate the file again below.
<-time.After(POLL_DURATION) <-time.After(watch.POLL_DURATION)
} }
// Delete after a reasonable delay, to give tail sufficient time // Delete after a reasonable delay, to give tail sufficient time
@ -206,7 +207,7 @@ func NewTailTest(name string, t *testing.T) TailTest {
} }
// Use a smaller poll duration for faster test runs. // Use a smaller poll duration for faster test runs.
POLL_DURATION = 25 * time.Millisecond watch.POLL_DURATION = 25 * time.Millisecond
return tt return tt
} }

215
watch.go
View File

@ -1,215 +0,0 @@
// Copyright (c) 2013 ActiveState Software Inc. All rights reserved.
package tail
import (
"github.com/howeyc/fsnotify"
"os"
"path/filepath"
"sync"
"time"
)
// FileWatcher monitors file-level events.
type FileWatcher interface {
// BlockUntilExists blocks until the missing file comes into
// existence. If the file already exists, returns immediately.
BlockUntilExists() error
// ChangeEvents returns a channel of events corresponding to the
// times the file is ready to be read.
ChangeEvents(os.FileInfo) chan bool
}
// InotifyFileWatcher uses inotify to monitor file changes.
type InotifyFileWatcher struct {
Filename string
Size int64
}
func NewInotifyFileWatcher(filename string) *InotifyFileWatcher {
fw := &InotifyFileWatcher{filename, 0}
return fw
}
func (fw *InotifyFileWatcher) BlockUntilExists() error {
w, err := fsnotify.NewWatcher()
if err != nil {
return err
}
defer w.Close()
dirname := filepath.Dir(fw.Filename)
// Watch for new files to be created in the parent directory.
err = w.WatchFlags(dirname, fsnotify.FSN_CREATE)
if err != nil {
return err
}
defer w.RemoveWatch(filepath.Dir(fw.Filename))
// Do a real check now as the file might have been created before
// calling `WatchFlags` above.
if _, err = os.Stat(fw.Filename); !os.IsNotExist(err) {
// file exists, or stat returned an error.
return err
}
for {
evt := <-w.Event
if evt.Name == fw.Filename {
break
}
}
return nil
}
// ChangeEvents returns a channel that gets updated when the file is ready to be read.
func (fw *InotifyFileWatcher) ChangeEvents(fi os.FileInfo) chan bool {
w, err := fsnotify.NewWatcher()
if err != nil {
panic(err)
}
err = w.Watch(fw.Filename)
if err != nil {
panic(err)
}
ch := make(chan bool)
fw.Size = fi.Size()
go func() {
defer w.Close()
defer w.RemoveWatch(fw.Filename)
defer close(ch)
for {
prevSize := fw.Size
evt := <-w.Event
switch {
case evt.IsDelete():
fallthrough
case evt.IsRename():
return
case evt.IsModify():
fi, err := os.Stat(fw.Filename)
if err != nil {
// XXX: no panic here
panic(err)
}
fw.Size = fi.Size()
if prevSize > 0 && prevSize > fw.Size {
return
}
// send only if channel is empty.
select {
case ch <- true:
default:
}
}
}
}()
return ch
}
// PollingFileWatcher polls the file for changes.
type PollingFileWatcher struct {
Filename string
Size int64
}
func NewPollingFileWatcher(filename string) *PollingFileWatcher {
fw := &PollingFileWatcher{filename, 0}
return fw
}
var POLL_DURATION time.Duration
func (fw *PollingFileWatcher) BlockUntilExists() error {
for {
if _, err := os.Stat(fw.Filename); err == nil {
return nil
} else if !os.IsNotExist(err) {
return err
}
time.Sleep(POLL_DURATION)
}
panic("unreachable")
}
func (fw *PollingFileWatcher) ChangeEvents(origFi os.FileInfo) chan bool {
ch := make(chan bool)
stop := make(chan bool)
var once sync.Once
var prevModTime time.Time
// XXX: use tomb.Tomb to cleanly manage these goroutines. replace
// the panic (below) with tomb's Kill.
stopAndClose := func() {
go func() {
close(ch)
stop <- true
}()
}
fw.Size = origFi.Size()
go func() {
prevSize := fw.Size
for {
select {
case <-stop:
return
default:
}
time.Sleep(POLL_DURATION)
fi, err := os.Stat(fw.Filename)
if err != nil {
if os.IsNotExist(err) {
once.Do(stopAndClose)
continue
}
/// XXX: do not panic here.
panic(err)
}
// File got moved/rename within POLL_DURATION?
if !os.SameFile(origFi, fi) {
once.Do(stopAndClose)
continue
}
// Was the file truncated?
fw.Size = fi.Size()
if prevSize > 0 && prevSize > fw.Size {
once.Do(stopAndClose)
continue
}
// If the file was changed since last check, notify.
modTime := fi.ModTime()
if modTime != prevModTime {
prevModTime = modTime
select {
case ch <- true:
default:
}
}
}
}()
return ch
}
func init() {
POLL_DURATION = 250 * time.Millisecond
}

107
watch/inotify.go Normal file
View File

@ -0,0 +1,107 @@
// Copyright (c) 2013 ActiveState Software Inc. All rights reserved.
package watch
import (
"github.com/howeyc/fsnotify"
"os"
"path/filepath"
)
// InotifyFileWatcher uses inotify to monitor file changes.
type InotifyFileWatcher struct {
Filename string
Size int64
}
func NewInotifyFileWatcher(filename string) *InotifyFileWatcher {
fw := &InotifyFileWatcher{filename, 0}
return fw
}
func (fw *InotifyFileWatcher) BlockUntilExists() error {
w, err := fsnotify.NewWatcher()
if err != nil {
return err
}
defer w.Close()
dirname := filepath.Dir(fw.Filename)
// Watch for new files to be created in the parent directory.
err = w.WatchFlags(dirname, fsnotify.FSN_CREATE)
if err != nil {
return err
}
defer w.RemoveWatch(filepath.Dir(fw.Filename))
// Do a real check now as the file might have been created before
// calling `WatchFlags` above.
if _, err = os.Stat(fw.Filename); !os.IsNotExist(err) {
// file exists, or stat returned an error.
return err
}
for {
evt := <-w.Event
if evt.Name == fw.Filename {
break
}
}
return nil
}
// ChangeEvents returns a channel that gets updated when the file is ready to be read.
func (fw *InotifyFileWatcher) ChangeEvents(fi os.FileInfo) chan bool {
w, err := fsnotify.NewWatcher()
if err != nil {
panic(err)
}
err = w.Watch(fw.Filename)
if err != nil {
panic(err)
}
ch := make(chan bool)
fw.Size = fi.Size()
go func() {
defer w.Close()
defer w.RemoveWatch(fw.Filename)
defer close(ch)
for {
prevSize := fw.Size
evt := <-w.Event
switch {
case evt.IsDelete():
fallthrough
case evt.IsRename():
return
case evt.IsModify():
fi, err := os.Stat(fw.Filename)
if err != nil {
// XXX: no panic here
panic(err)
}
fw.Size = fi.Size()
if prevSize > 0 && prevSize > fw.Size {
return
}
// send only if channel is empty.
select {
case ch <- true:
default:
}
}
}
}()
return ch
}

104
watch/polling.go Normal file
View File

@ -0,0 +1,104 @@
// Copyright (c) 2013 ActiveState Software Inc. All rights reserved.
package watch
import (
"os"
"sync"
"time"
)
// PollingFileWatcher polls the file for changes.
type PollingFileWatcher struct {
Filename string
Size int64
}
func NewPollingFileWatcher(filename string) *PollingFileWatcher {
fw := &PollingFileWatcher{filename, 0}
return fw
}
var POLL_DURATION time.Duration
func (fw *PollingFileWatcher) BlockUntilExists() error {
for {
if _, err := os.Stat(fw.Filename); err == nil {
return nil
} else if !os.IsNotExist(err) {
return err
}
time.Sleep(POLL_DURATION)
}
panic("unreachable")
}
func (fw *PollingFileWatcher) ChangeEvents(origFi os.FileInfo) chan bool {
ch := make(chan bool)
stop := make(chan bool)
var once sync.Once
var prevModTime time.Time
// XXX: use tomb.Tomb to cleanly manage these goroutines. replace
// the panic (below) with tomb's Kill.
stopAndClose := func() {
go func() {
close(ch)
stop <- true
}()
}
fw.Size = origFi.Size()
go func() {
prevSize := fw.Size
for {
select {
case <-stop:
return
default:
}
time.Sleep(POLL_DURATION)
fi, err := os.Stat(fw.Filename)
if err != nil {
if os.IsNotExist(err) {
once.Do(stopAndClose)
continue
}
/// XXX: do not panic here.
panic(err)
}
// File got moved/rename within POLL_DURATION?
if !os.SameFile(origFi, fi) {
once.Do(stopAndClose)
continue
}
// Was the file truncated?
fw.Size = fi.Size()
if prevSize > 0 && prevSize > fw.Size {
once.Do(stopAndClose)
continue
}
// If the file was changed since last check, notify.
modTime := fi.ModTime()
if modTime != prevModTime {
prevModTime = modTime
select {
case ch <- true:
default:
}
}
}
}()
return ch
}
func init() {
POLL_DURATION = 250 * time.Millisecond
}

19
watch/watch.go Normal file
View File

@ -0,0 +1,19 @@
// Copyright (c) 2013 ActiveState Software Inc. All rights reserved.
package watch
import (
"os"
)
// FileWatcher monitors file-level events.
type FileWatcher interface {
// BlockUntilExists blocks until the missing file comes into
// existence. If the file already exists, returns immediately.
BlockUntilExists() error
// ChangeEvents returns a channel of events corresponding to the
// times the file is ready to be read.
ChangeEvents(os.FileInfo) chan bool
}