FS watcher failures cancel all log watches

internal/version/VERSION

@@ -1 +1 @@
-v0.4.0
+v0.4.1

logfile.go

@@ -75,7 +75,7 @@ func newLogFile(fp string, bs int) (*logFile, error) {
 // If the file does not exist at startup, the data processor will wait for the file to be created before continuing.
 //
 // The data processor expects the log data to be newline delimited
-func (lf *logFile) dataProcessor(ctx context.Context, ewCancelChan <-chan error, stopChan chan<- string) {
+func (lf *logFile) dataProcessor(ctx context.Context, ewCancelChan <-chan struct{}, stopChan chan<- string) {
  go func() {
  defer close(lf.dataChan)
  defer close(lf.errorChan)
@@ -181,7 +181,7 @@ func (lf *logFile) dataProcessor(ctx context.Context, ewCancelChan <-chan error,
  continue readLoop
  }
  continue
- case err := <-ewCancelChan:
+ case <-ewCancelChan:
  stopChan <- lf.filePath
  lf.errorChan <- NewLogWhaleError(ErrorStateCancelled, fmt.Sprintf("operation cancelled"), err)
  return

logmanager.go

@@ -29,7 +29,7 @@ type LogManager struct {
  ctx context.Context
  ctxCancel context.CancelFunc
 
- evwCancelChan chan error // event watcher cancel channel
+ evwCancelChan chan struct{} // event watcher cancel channel
  fileWatcher *fsnotify.Watcher
 
  lwMutex *sync.RWMutex // lwMutex is a mutex for the logsWatched map
@@ -65,7 +65,7 @@ func NewLogManager(ctx context.Context, options ...Option) (*LogManager, error)
  pathsWatched: make(map[string]int),
  pwMutex: &sync.RWMutex{},
 
- evwCancelChan: make(chan error),
+ evwCancelChan: make(chan struct{}),
  removeChan: make(chan string),
  }
 
@@ -224,7 +224,7 @@ func (lm *LogManager) eventWatcher() {
  return
  case fse, ok := <-lm.fileWatcher.Events:
  if !ok {
- lm.evwCancelChan <- fmt.Errorf("file watcher closed unexpectedly")
+ close(lm.evwCancelChan)
  return
  }
 
@@ -268,8 +268,8 @@ func (lm *LogManager) eventWatcher() {
  lm.lwMutex.RUnlock()
  continue
  }
- case err := <-lm.fileWatcher.Errors:
- lm.evwCancelChan <- err
+ case <-lm.fileWatcher.Errors:
+ close(lm.evwCancelChan)
  return
  }
  }