src/foundation/api/revel/watcher.go

   1 // Copyright (c) 2012-2016 The Revel Framework Authors, All rights reserved.
   2 // Revel Framework source code and usage is governed by a MIT style
   3 // license that can be found in the LICENSE file.
   4
   5 package revel
   6
   7 import (
   8         "os"
   9         "path/filepath"
  10         "strings"
  11         "sync"
  12
  13         "gopkg.in/fsnotify/fsnotify.v1"
  14         "time"
  15 )
  16
  17 // Listener is an interface for receivers of filesystem events.
  18 type Listener interface {
  19         // Refresh is invoked by the watcher on relevant filesystem events.
  20         // If the listener returns an error, it is served to the user on the current request.
  21         Refresh() *Error
  22 }
  23
  24 // DiscerningListener allows the receiver to selectively watch files.
  25 type DiscerningListener interface {
  26         Listener
  27         WatchDir(info os.FileInfo) bool
  28         WatchFile(basename string) bool
  29 }
  30
  31 // Watcher allows listeners to register to be notified of changes under a given
  32 // directory.
  33 type Watcher struct {
  34         serial              bool                // true to process events in serial
  35         watchers            []*fsnotify.Watcher // Parallel arrays of watcher/listener pairs.
  36         listeners           []Listener          // List of listeners for watcher
  37         forceRefresh        bool                // True to force the refresh
  38         lastError           int                 // The last error found
  39         notifyMutex         sync.Mutex          // The mutext to serialize watches
  40         refreshTimer        *time.Timer         // The timer to countdown the next refresh
  41         timerMutex          *sync.Mutex         // A mutex to prevent concurrent updates
  42         refreshChannel      chan *Error         // The error channel to listen to when waiting for a refresh
  43         refreshChannelCount int                 // The number of clients listening on the channel
  44         refreshTimerMS      time.Duration       // The number of milliseconds between refreshing builds
  45 }
  46
  47 func NewWatcher() *Watcher {
  48         return &Watcher{
  49                 forceRefresh:        true,
  50                 lastError:           -1,
  51                 refreshTimerMS:      time.Duration(Config.IntDefault("watch.rebuild.delay", 10)),
  52                 timerMutex:          &sync.Mutex{},
  53                 refreshChannel:      make(chan *Error, 10),
  54                 refreshChannelCount: 0,
  55         }
  56 }
  57
  58 // Listen registers for events within the given root directories (recursively).
  59 func (w *Watcher) Listen(listener Listener, roots ...string) {
  60         watcher, err := fsnotify.NewWatcher()
  61         if err != nil {
  62                 utilLog.Fatal("Watcher: Failed to create watcher", "error", err)
  63         }
  64
  65         // Replace the unbuffered Event channel with a buffered one.
  66         // Otherwise multiple change events only come out one at a time, across
  67         // multiple page views.  (There appears no way to "pump" the events out of
  68         // the watcher)
  69         // This causes a notification when you do a check in go, since you are modifying a buffer in use
  70         watcher.Events = make(chan fsnotify.Event, 100)
  71         watcher.Errors = make(chan error, 10)
  72
  73         // Walk through all files / directories under the root, adding each to watcher.
  74         for _, p := range roots {
  75                 // is the directory / file a symlink?
  76                 f, err := os.Lstat(p)
  77                 if err == nil && f.Mode()&os.ModeSymlink == os.ModeSymlink {
  78                         var realPath string
  79                         realPath, err = filepath.EvalSymlinks(p)
  80                         if err != nil {
  81                                 panic(err)
  82                         }
  83                         p = realPath
  84                 }
  85
  86                 fi, err := os.Stat(p)
  87                 if err != nil {
  88                         utilLog.Error("Watcher: Failed to stat watched path, code will continue but auto updates will not work", "path", p, "error", err)
  89                         continue
  90                 }
  91
  92                 // If it is a file, watch that specific file.
  93                 if !fi.IsDir() {
  94                         err = watcher.Add(p)
  95                         if err != nil {
  96                                 utilLog.Error("Watcher: Failed to watch, code will continue but auto updates will not work", "path", p, "error", err)
  97                         }
  98                         continue
  99                 }
 100
 101                 var watcherWalker func(path string, info os.FileInfo, err error) error
 102
 103                 watcherWalker = func(path string, info os.FileInfo, err error) error {
 104                         if err != nil {
 105                                 utilLog.Error("Watcher: Error walking path:", "error", err)
 106                                 return nil
 107                         }
 108
 109                         if info.IsDir() {
 110                                 if dl, ok := listener.(DiscerningListener); ok {
 111                                         if !dl.WatchDir(info) {
 112                                                 return filepath.SkipDir
 113                                         }
 114                                 }
 115
 116                                 err := watcher.Add(path)
 117                                 if err != nil {
 118                                         utilLog.Error("Watcher: Failed to watch this path, code will continue but auto updates will not work", "path", path, "error", err)
 119                                 }
 120                         }
 121                         return nil
 122                 }
 123
 124                 // Else, walk the directory tree.
 125                 err = Walk(p, watcherWalker)
 126                 if err != nil {
 127                         utilLog.Error("Watcher: Failed to walk directory, code will continue but auto updates will not work", "path", p, "error", err)
 128                 }
 129         }
 130
 131         if w.eagerRebuildEnabled() {
 132                 // Create goroutine to notify file changes in real time
 133                 go w.NotifyWhenUpdated(listener, watcher)
 134         }
 135
 136         w.watchers = append(w.watchers, watcher)
 137         w.listeners = append(w.listeners, listener)
 138 }
 139
 140 // NotifyWhenUpdated notifies the watcher when a file event is received.
 141 func (w *Watcher) NotifyWhenUpdated(listener Listener, watcher *fsnotify.Watcher) {
 142
 143         for {
 144                 select {
 145                 case ev := <-watcher.Events:
 146                         if w.rebuildRequired(ev, listener) {
 147                                 // Serialize listener.Refresh() calls.
 148                                 if w.serial {
 149                                         // Serialize listener.Refresh() calls.
 150                                         w.notifyMutex.Lock()
 151
 152                                         if err := listener.Refresh(); err != nil {
 153                                                 utilLog.Error("Watcher: Listener refresh reported error:", "error", err)
 154                                         }
 155                                         w.notifyMutex.Unlock()
 156                                 } else {
 157                                         // Run refresh in parallel
 158                                         go func() {
 159                                                 w.notifyInProcess(listener)
 160                                         }()
 161                                 }
 162                         }
 163                 case <-watcher.Errors:
 164                         continue
 165                 }
 166         }
 167 }
 168
 169 // Notify causes the watcher to forward any change events to listeners.
 170 // It returns the first (if any) error returned.
 171 func (w *Watcher) Notify() *Error {
 172         // Serialize Notify() calls.
 173         w.notifyMutex.Lock()
 174         defer w.notifyMutex.Unlock()
 175
 176         for i, watcher := range w.watchers {
 177                 listener := w.listeners[i]
 178
 179                 // Pull all pending events / errors from the watcher.
 180                 refresh := false
 181                 for {
 182                         select {
 183                         case ev := <-watcher.Events:
 184                                 if w.rebuildRequired(ev, listener) {
 185                                         refresh = true
 186                                 }
 187                                 continue
 188                         case <-watcher.Errors:
 189                                 continue
 190                         default:
 191                                 // No events left to pull
 192                         }
 193                         break
 194                 }
 195
 196                 if w.forceRefresh || refresh || w.lastError == i {
 197                         var err *Error
 198                         if w.serial {
 199                                 err = listener.Refresh()
 200                         } else {
 201                                 err = w.notifyInProcess(listener)
 202                         }
 203
 204                         if err != nil {
 205                                 w.lastError = i
 206                                 return err
 207                         }
 208                 }
 209         }
 210
 211         w.forceRefresh = false
 212         w.lastError = -1
 213         return nil
 214 }
 215
 216 // Build a queue for refresh notifications
 217 // this will not return until one of the queue completes
 218 func (w *Watcher) notifyInProcess(listener Listener) (err *Error) {
 219         shouldReturn := false
 220         // This code block ensures that either a timer is created
 221         // or that a process would be added the the h.refreshChannel
 222         func() {
 223                 w.timerMutex.Lock()
 224                 defer w.timerMutex.Unlock()
 225                 // If we are in the process of a rebuild, forceRefresh will always be true
 226                 w.forceRefresh = true
 227                 if w.refreshTimer != nil {
 228                         utilLog.Info("Found existing timer running, resetting")
 229                         w.refreshTimer.Reset(time.Millisecond * w.refreshTimerMS)
 230                         shouldReturn = true
 231                         w.refreshChannelCount++
 232                 } else {
 233                         w.refreshTimer = time.NewTimer(time.Millisecond * w.refreshTimerMS)
 234                 }
 235         }()
 236
 237         // If another process is already waiting for the timer this one
 238         // only needs to return the output from the channel
 239         if shouldReturn {
 240                 return <-w.refreshChannel
 241         }
 242         utilLog.Info("Waiting for refresh timer to expire")
 243         <-w.refreshTimer.C
 244         w.timerMutex.Lock()
 245
 246         // Ensure the queue is properly dispatched even if a panic occurs
 247         defer func() {
 248                 for x := 0; x < w.refreshChannelCount; x++ {
 249                         w.refreshChannel <- err
 250                 }
 251                 w.refreshChannelCount = 0
 252                 w.refreshTimer = nil
 253                 w.timerMutex.Unlock()
 254         }()
 255
 256         err = listener.Refresh()
 257         if err != nil {
 258                 utilLog.Info("Watcher: Recording error last build, setting rebuild on", "error", err)
 259         } else {
 260                 w.lastError = -1
 261                 w.forceRefresh = false
 262         }
 263         utilLog.Info("Rebuilt, result", "error", err)
 264         return
 265 }
 266
 267 // If watch.mode is set to eager, the application is rebuilt immediately
 268 // when a source file is changed.
 269 // This feature is available only in dev mode.
 270 func (w *Watcher) eagerRebuildEnabled() bool {
 271         return Config.BoolDefault("mode.dev", true) &&
 272                 Config.BoolDefault("watch", true) &&
 273                 Config.StringDefault("watch.mode", "normal") == "eager"
 274 }
 275
 276 func (w *Watcher) rebuildRequired(ev fsnotify.Event, listener Listener) bool {
 277         // Ignore changes to dotfiles.
 278         if strings.HasPrefix(filepath.Base(ev.Name), ".") {
 279                 return false
 280         }
 281
 282         if dl, ok := listener.(DiscerningListener); ok {
 283                 if !dl.WatchFile(ev.Name) || ev.Op&fsnotify.Chmod == fsnotify.Chmod {
 284                         return false
 285                 }
 286         }
 287         return true
 288 }
 289
 290 var WatchFilter = func(c *Controller, fc []Filter) {
 291         if MainWatcher != nil {
 292                 err := MainWatcher.Notify()
 293                 if err != nil {
 294                         c.Result = c.RenderError(err)
 295                         return
 296                 }
 297         }
 298         fc[0](c, fc[1:])
 299 }