poller.go 4.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204
  1. package filenotify
  2. import (
  3. "errors"
  4. "fmt"
  5. "os"
  6. "sync"
  7. "time"
  8. "github.com/Sirupsen/logrus"
  9. "github.com/fsnotify/fsnotify"
  10. )
  11. var (
  12. // errPollerClosed is returned when the poller is closed
  13. errPollerClosed = errors.New("poller is closed")
  14. // errNoSuchWatch is returned when trying to remove a watch that doesn't exist
  15. errNoSuchWatch = errors.New("watch does not exist")
  16. )
  17. // watchWaitTime is the time to wait between file poll loops
  18. const watchWaitTime = 200 * time.Millisecond
  19. // filePoller is used to poll files for changes, especially in cases where fsnotify
  20. // can't be run (e.g. when inotify handles are exhausted)
  21. // filePoller satisfies the FileWatcher interface
  22. type filePoller struct {
  23. // watches is the list of files currently being polled, close the associated channel to stop the watch
  24. watches map[string]chan struct{}
  25. // events is the channel to listen to for watch events
  26. events chan fsnotify.Event
  27. // errors is the channel to listen to for watch errors
  28. errors chan error
  29. // mu locks the poller for modification
  30. mu sync.Mutex
  31. // closed is used to specify when the poller has already closed
  32. closed bool
  33. }
  34. // Add adds a filename to the list of watches
  35. // once added the file is polled for changes in a separate goroutine
  36. func (w *filePoller) Add(name string) error {
  37. w.mu.Lock()
  38. defer w.mu.Unlock()
  39. if w.closed {
  40. return errPollerClosed
  41. }
  42. f, err := os.Open(name)
  43. if err != nil {
  44. return err
  45. }
  46. fi, err := os.Stat(name)
  47. if err != nil {
  48. return err
  49. }
  50. if w.watches == nil {
  51. w.watches = make(map[string]chan struct{})
  52. }
  53. if _, exists := w.watches[name]; exists {
  54. return fmt.Errorf("watch exists")
  55. }
  56. chClose := make(chan struct{})
  57. w.watches[name] = chClose
  58. go w.watch(f, fi, chClose)
  59. return nil
  60. }
  61. // Remove stops and removes watch with the specified name
  62. func (w *filePoller) Remove(name string) error {
  63. w.mu.Lock()
  64. defer w.mu.Unlock()
  65. return w.remove(name)
  66. }
  67. func (w *filePoller) remove(name string) error {
  68. if w.closed {
  69. return errPollerClosed
  70. }
  71. chClose, exists := w.watches[name]
  72. if !exists {
  73. return errNoSuchWatch
  74. }
  75. close(chClose)
  76. delete(w.watches, name)
  77. return nil
  78. }
  79. // Events returns the event channel
  80. // This is used for notifications on events about watched files
  81. func (w *filePoller) Events() <-chan fsnotify.Event {
  82. return w.events
  83. }
  84. // Errors returns the errors channel
  85. // This is used for notifications about errors on watched files
  86. func (w *filePoller) Errors() <-chan error {
  87. return w.errors
  88. }
  89. // Close closes the poller
  90. // All watches are stopped, removed, and the poller cannot be added to
  91. func (w *filePoller) Close() error {
  92. w.mu.Lock()
  93. defer w.mu.Unlock()
  94. if w.closed {
  95. return nil
  96. }
  97. w.closed = true
  98. for name := range w.watches {
  99. w.remove(name)
  100. delete(w.watches, name)
  101. }
  102. return nil
  103. }
  104. // sendEvent publishes the specified event to the events channel
  105. func (w *filePoller) sendEvent(e fsnotify.Event, chClose <-chan struct{}) error {
  106. select {
  107. case w.events <- e:
  108. case <-chClose:
  109. return fmt.Errorf("closed")
  110. }
  111. return nil
  112. }
  113. // sendErr publishes the specified error to the errors channel
  114. func (w *filePoller) sendErr(e error, chClose <-chan struct{}) error {
  115. select {
  116. case w.errors <- e:
  117. case <-chClose:
  118. return fmt.Errorf("closed")
  119. }
  120. return nil
  121. }
  122. // watch is responsible for polling the specified file for changes
  123. // upon finding changes to a file or errors, sendEvent/sendErr is called
  124. func (w *filePoller) watch(f *os.File, lastFi os.FileInfo, chClose chan struct{}) {
  125. defer f.Close()
  126. for {
  127. time.Sleep(watchWaitTime)
  128. select {
  129. case <-chClose:
  130. logrus.Debugf("watch for %s closed", f.Name())
  131. return
  132. default:
  133. }
  134. fi, err := os.Stat(f.Name())
  135. if err != nil {
  136. // if we got an error here and lastFi is not set, we can presume that nothing has changed
  137. // This should be safe since before `watch()` is called, a stat is performed, there is any error `watch` is not called
  138. if lastFi == nil {
  139. continue
  140. }
  141. // If it doesn't exist at this point, it must have been removed
  142. // no need to send the error here since this is a valid operation
  143. if os.IsNotExist(err) {
  144. if err := w.sendEvent(fsnotify.Event{Op: fsnotify.Remove, Name: f.Name()}, chClose); err != nil {
  145. return
  146. }
  147. lastFi = nil
  148. continue
  149. }
  150. // at this point, send the error
  151. if err := w.sendErr(err, chClose); err != nil {
  152. return
  153. }
  154. continue
  155. }
  156. if lastFi == nil {
  157. if err := w.sendEvent(fsnotify.Event{Op: fsnotify.Create, Name: fi.Name()}, chClose); err != nil {
  158. return
  159. }
  160. lastFi = fi
  161. continue
  162. }
  163. if fi.Mode() != lastFi.Mode() {
  164. if err := w.sendEvent(fsnotify.Event{Op: fsnotify.Chmod, Name: fi.Name()}, chClose); err != nil {
  165. return
  166. }
  167. lastFi = fi
  168. continue
  169. }
  170. if fi.ModTime() != lastFi.ModTime() || fi.Size() != lastFi.Size() {
  171. if err := w.sendEvent(fsnotify.Event{Op: fsnotify.Write, Name: fi.Name()}, chClose); err != nil {
  172. return
  173. }
  174. lastFi = fi
  175. continue
  176. }
  177. }
  178. }