#5 prepare polling watcher with multiple logfile support

Author: fstab

tailer/fswatcher/fswatcher.go

@@ -22,6 +22,7 @@ import (
 	"io"
 	"os"
 	"path/filepath"
+	"time"
 )
 
 type FileTailer interface {
@@ -92,6 +93,17 @@ func (t *fileTailer) Close() {
 }
 
 func RunFileTailer(globs []glob.Glob, readall bool, failOnMissingFile bool, log logrus.FieldLogger) (FileTailer, error) {
+	return runFileTailer(initWatcher, globs, readall, failOnMissingFile, log)
+}
+
+func RunPollingFileTailer(globs []glob.Glob, readall bool, failOnMissingFile bool, pollInterval time.Duration, log logrus.FieldLogger) (FileTailer, error) {
+	initFunc := func() (fswatcher, Error) {
+		return initPollingWatcher(pollInterval)
+	}
+	return runFileTailer(initFunc, globs, readall, failOnMissingFile, log)
+}
+
+func runFileTailer(initFunc func() (fswatcher, Error), globs []glob.Glob, readall bool, failOnMissingFile bool, log logrus.FieldLogger) (FileTailer, error) {
 
 	var (
 		t   *fileTailer
@@ -106,7 +118,7 @@ func RunFileTailer(globs []glob.Glob, readall bool, failOnMissingFile bool, log
 		done:         make(chan struct{}),
 	}
 
-	t.osSpecific, Err = initWatcher()
+	t.osSpecific, Err = initFunc()
 	if Err != nil {
 		return nil, Err
 	}
@@ -127,10 +139,10 @@ func RunFileTailer(globs []glob.Glob, readall bool, failOnMissingFile bool, log
 		eventProducerLoop := t.osSpecific.runFseventProducerLoop()
 		defer eventProducerLoop.Close()
 
-		for _, dirPath := range t.watchedDirs {
-			dirLogger := log.WithField("directory", dirPath)
+		for _, dir := range t.watchedDirs {
+			dirLogger := log.WithField("directory", dir.Path())
 			dirLogger.Debugf("initializing directory")
-			Err = t.syncFilesInDir(dirPath, readall, dirLogger)
+			Err = t.syncFilesInDir(dir, readall, dirLogger)
 			if Err != nil {
 				select {
 				case <-t.done:

tailer/fswatcher/fswatcher_darwin.go

@@ -65,19 +65,32 @@ func initWatcher() (fswatcher, Error) {
 
 func (w *watcher) watchDir(path string) (*Dir, Error) {
 	var (
+		dir         *Dir
 		err         error
-		dir         *os.File
+		Err         Error
 		zeroTimeout = syscall.NsecToTimespec(0) // timeout zero means non-blocking kevent() call
 	)
-	dir, err = os.Open(path)
-	if err != nil {
-		return nil, NewErrorf(NotSpecified, err, "%v: open() failed", path)
+	dir, Err = newDir(path)
+	if Err != nil {
+		return nil, Err
 	}
-	_, err = syscall.Kevent(w.kq, []syscall.Kevent_t{makeEvent(dir)}, nil, &zeroTimeout)
+	_, err = syscall.Kevent(w.kq, []syscall.Kevent_t{makeEvent(dir.file)}, nil, &zeroTimeout)
 	if err != nil {
-		dir.Close()
+		dir.file.Close()
 		return nil, NewErrorf(NotSpecified, err, "%v: kevent() failed", path)
 	}
+	return dir, nil
+}
+
+func newDir(path string) (*Dir, Error) {
+	var (
+		err error
+		dir *os.File
+	)
+	dir, err = os.Open(path)
+	if err != nil {
+		return nil, NewErrorf(NotSpecified, err, "%v: open() failed", path)
+	}
 	return &Dir{dir}, nil
 }
 

tailer/fswatcher/fswatcher_linux.go

@@ -76,14 +76,23 @@ func initWatcher() (fswatcher, Error) {
 
 func (w *watcher) watchDir(path string) (*Dir, Error) {
 	var (
-		wd  int
+		dir *Dir
 		err error
+		Err Error
 	)
-	wd, err = syscall.InotifyAddWatch(w.fd, path, syscall.IN_MODIFY|syscall.IN_MOVED_FROM|syscall.IN_MOVED_TO|syscall.IN_DELETE|syscall.IN_CREATE)
+	dir, Err = newDir(path)
+	if Err != nil {
+		return nil, Err
+	}
+	dir.wd, err = syscall.InotifyAddWatch(w.fd, path, syscall.IN_MODIFY|syscall.IN_MOVED_FROM|syscall.IN_MOVED_TO|syscall.IN_DELETE|syscall.IN_CREATE)
 	if err != nil {
 		return nil, NewErrorf(NotSpecified, err, "%q: inotify_add_watch() failed", path)
 	}
-	return &Dir{wd: wd, path: path}, nil
+	return dir, nil
+}
+
+func newDir(path string) (*Dir, Error) {
+	return &Dir{path: path}, nil
 }
 
 func (w *watcher) watchFile(_ fileMeta) Error {

tailer/fswatcher/fswatcher_windows.go

@@ -73,10 +73,23 @@ func initWatcher() (fswatcher, Error) {
 }
 
 func (w *watcher) watchDir(path string) (*Dir, Error) {
-	err := w.winWatcher.Watch(path)
+	var (
+		dir *Dir
+		err error
+		Err Error
+	)
+	dir, Err = newDir(path)
+	if Err != nil {
+		return nil, Err
+	}
+	err = w.winWatcher.Watch(path)
 	if err != nil {
 		return nil, NewErrorf(NotSpecified, err, "%v: failed to watch directory", path)
 	}
+	return dir, nil
+}
+
+func newDir(path string) (*Dir, Error) {
 	return &Dir{path: path}, nil
 }
 

tailer/fswatcher/pollingFseventProducerLoop.go

@@ -0,0 +1,49 @@
+package fswatcher
+
+import "time"
+
+type pollloop struct {
+	events chan fsevent
+	errors chan Error // unused
+	done   chan struct{}
+}
+
+func (l *pollloop) Events() chan fsevent {
+	return l.events
+}
+
+func (l *pollloop) Errors() chan Error {
+	return l.errors
+}
+
+func (l *pollloop) Close() {
+	close(l.done)
+}
+
+func runPollLoop(pollInterval time.Duration) *pollloop {
+
+	events := make(chan fsevent)
+	errors := make(chan Error) // unused
+	done := make(chan struct{})
+
+	go func() {
+		defer func() {
+			close(events)
+			close(errors)
+		}()
+		for {
+			tick := time.After(pollInterval)
+			select {
+			case <-tick:
+				events <- struct{}{}
+			case <-done:
+				return
+			}
+		}
+	}()
+	return &pollloop{
+		events: events,
+		errors: errors,
+		done:   done,
+	}
+}

tailer/fswatcher/pollingwatcher.go

@@ -0,0 +1,46 @@
+package fswatcher
+
+import (
+	"github.com/sirupsen/logrus"
+	"time"
+)
+
+type pollingWatcher struct {
+	pollInterval time.Duration
+}
+
+func initPollingWatcher(pollInterval time.Duration) (fswatcher, Error) {
+	return &pollingWatcher{
+		pollInterval: pollInterval,
+	}, nil
+}
+
+func (w *pollingWatcher) runFseventProducerLoop() fseventProducerLoop {
+	return runPollLoop(w.pollInterval)
+}
+
+func (w *pollingWatcher) processEvent(t *fileTailer, fsevent fsevent, log logrus.FieldLogger) Error {
+	for _, dir := range t.watchedDirs {
+		err := t.syncFilesInDir(dir, true, log)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (w *pollingWatcher) Close() error {
+	return nil
+}
+
+func (w *pollingWatcher) watchDir(path string) (*Dir, Error) {
+	return newDir(path)
+}
+
+func (w *pollingWatcher) unwatchDir(dir *Dir) error {
+	return nil
+}
+
+func (w *pollingWatcher) watchFile(file fileMeta) Error {
+	return nil
+}