pkg/watcher/watcher.go
Rama main · réplica de lectura del repositorio original.
package watcher
import (
"os"
"path/filepath"
"strings"
"sync"
"time"
"github.com/bep/debounce"
"github.com/fsnotify/fsnotify"
"github.com/rs/zerolog"
"github.com/rs/zerolog/log"
"encr.dev/pkg/eerror"
)
type Watcher struct {
mutex sync.Mutex
eventCond *sync.Cond
events *Events
signalDebounce func(func())
log *zerolog.Logger
appRoot string
watcher *fsnotify.Watcher
directories map[string]struct{}
stop chan struct{}
closeOnce sync.Once
closed bool // guarded by mutex
}
func New(appID string) (*Watcher, error) {
fswatcher, err := fsnotify.NewWatcher()
if err != nil {
return nil, eerror.Wrap(err, "watcher", "unable to create watcher", map[string]interface{}{"app": appID})
}
logger := log.With().Str("component", "watcher").Str("app", appID).Logger()
logger.Debug().Msg("File system watcher created")
w := &Watcher{
watcher: fswatcher,
log: &logger,
directories: make(map[string]struct{}),
stop: make(chan struct{}),
events: nil,
signalDebounce: debounce.New(50 * time.Millisecond),
}
w.eventCond = sync.NewCond(&w.mutex)
go w.listenForChangeEvents()
return w, nil
}
func (w *Watcher) RecursivelyWatch(folder string) error {
return filepath.WalkDir(folder, func(path string, info os.DirEntry, err error) error {
if err != nil {
return eerror.Wrap(err, "watcher", "unable to walk directory", map[string]any{"path": path})
}
if info.IsDir() {
folder := filepath.Clean(path)
if IgnoreFolder(folder) {
return filepath.SkipDir
}
// Track the fact we're watching this directory
w.mutex.Lock()
if w.closed {
w.mutex.Unlock()
return filepath.SkipAll
}
if _, found := w.directories[folder]; found {
w.mutex.Unlock()
return filepath.SkipDir
}
w.directories[folder] = struct{}{}
w.mutex.Unlock() // unlock here to prevent reentrant locks during recursion
// Now start watching this folder
addErr := w.watcher.Add(folder)
// Close may have run its removal sweep while the add was in
// flight, in which case this watch missed it and must be
// dropped here instead.
w.mutex.Lock()
stale := w.closed || addErr != nil
if stale {
delete(w.directories, folder)
}
w.mutex.Unlock()
if stale {
_ = w.watcher.Remove(folder)
if addErr != nil {
return eerror.Wrap(addErr, "watcher", "unable to add folder to watch", map[string]any{"folder": folder})
}
return filepath.SkipAll
}
}
return nil
})
}
func (w *Watcher) listenForChangeEvents() {
for {
select {
case <-w.stop:
_ = w.watcher.Close()
return
case event := <-w.watcher.Events:
if event.Has(fsnotify.Remove) {
w.handleDeleteEvent(event.Name)
} else if event.Has(fsnotify.Create) {
w.handleCreateEvent(event.Name)
} else if event.Has(fsnotify.Write) {
w.handleWriteEvent(event.Name)
}
case err := <-w.watcher.Errors:
w.log.Err(err).Msg("Watcher error")
}
}
}
func (w *Watcher) handleCreateEvent(path string) {
if info, err := os.Stat(path); err != nil {
w.log.Err(err).Str("path", path).Msg("unable to stat file")
} else if info.IsDir() {
if err := w.RecursivelyWatch(path); err != nil {
w.log.Err(err).Str("path", path).Msg("unable to start watching new directory")
}
w.recordExistingFiles(path)
} else {
w.recordEventInBatch(path, CREATED, info)
}
}
// recordExistingFiles records a CREATED event for each file in a newly created
// directory. Files created before the directory was watched
// (e.g. `mkdir -p dir && touch dir/file`) produce no events of their own.
func (w *Watcher) recordExistingFiles(dir string) {
_ = filepath.WalkDir(dir, func(path string, d os.DirEntry, err error) error {
if err != nil {
return nil
}
if d.IsDir() {
if IgnoreFolder(path) {
return filepath.SkipDir
}
return nil
}
if info, err := d.Info(); err == nil {
w.recordEventInBatch(path, CREATED, info)
}
return nil
})
}
func (w *Watcher) handleDeleteEvent(path string) {
path = filepath.Clean(path)
pathWithSep := path + string(filepath.Separator)
// If it's a directory we're watching, stop watching it
w.mutex.Lock()
for watchedFolder := range w.directories {
// I sthis the path itself, or a subdirectory thereof?
if strings.HasPrefix(watchedFolder, pathWithSep) || watchedFolder == path {
if err := w.watcher.Remove(watchedFolder); err != nil {
w.log.Err(err).Str("path", watchedFolder).Msg("unable to stop watching deleted directory")
}
delete(w.directories, watchedFolder)
}
}
w.mutex.Unlock()
w.recordEventInBatch(path, DELETED, nil)
}
func (w *Watcher) handleWriteEvent(path string) {
if info, err := os.Stat(path); err != nil {
w.log.Err(err).Str("path", path).Msg("unable to stat file")
} else if !info.IsDir() {
w.recordEventInBatch(path, MODIFIED, info)
}
}
func (w *Watcher) recordEventInBatch(path string, event EventType, info os.FileInfo) {
w.mutex.Lock()
defer w.mutex.Unlock()
if w.events == nil {
w.events = newEventBatch()
}
w.events.addEvent(path, event, info)
// Debounce the signal to avoid waking up on every event in case of a burst of events.
w.signalDebounce(func() {
w.eventCond.Signal()
})
}
func (w *Watcher) WaitForEvents() (events []Event, ok bool) {
w.eventCond.L.Lock()
defer w.eventCond.L.Unlock()
for {
select {
case <-w.stop:
// We're shutting down, so return immediately.
return nil, false
default:
}
if w.events == nil || len(w.events.latestEvents) == 0 {
// A wakeup doesn't guarantee events: Close broadcasts too,
// so re-check both the stop channel and the batch.
w.eventCond.Wait()
continue
}
events := w.events.Events()
w.events = newEventBatch()
return events, true
}
}
func (w *Watcher) Close() error {
w.closeOnce.Do(func() {
// Drop the watches ourselves before closing. fsnotify's kqueue
// backend (macOS/BSD) marks the watcher closed before removing its
// own watches, so every Remove it performs during Close is a no-op
// and each watched file and directory leaks its file descriptor.
// Removing them here, while the watcher is still open, actually
// releases them.
//
// Remove without holding the mutex: on some backends Remove waits
// on the goroutine that delivers events, which may itself be
// waiting on the mutex.
w.mutex.Lock()
// Stops RecursivelyWatch from registering new watches after the
// sweep below, which would leak them.
w.closed = true
folders := make([]string, 0, len(w.directories))
for folder := range w.directories {
folders = append(folders, folder)
}
clear(w.directories)
w.mutex.Unlock()
for _, folder := range folders {
_ = w.watcher.Remove(folder)
}
// Close under the mutex guarding the condition, so a WaitForEvents
// caller cannot check w.stop, miss the broadcast below, and then
// park on the condition forever.
w.mutex.Lock()
close(w.stop)
w.mutex.Unlock()
// Wake anyone parked in WaitForEvents so they observe the close.
w.eventCond.Broadcast()
})
return nil
}
func (w *Watcher) Done() <-chan struct{} {
return w.stop
}