OLD | NEW |
(Empty) | |
| 1 // Copyright (c) 2013, the Dart project authors. Please see the AUTHORS file |
| 2 // for details. All rights reserved. Use of this source code is governed by a |
| 3 // BSD-style license that can be found in the LICENSE file. |
| 4 |
| 5 library watcher.directory_watcher.linux; |
| 6 |
| 7 import 'dart:async'; |
| 8 import 'dart:io'; |
| 9 |
| 10 import '../directory_watcher.dart'; |
| 11 import '../utils.dart'; |
| 12 import '../watch_event.dart'; |
| 13 import 'resubscribable.dart'; |
| 14 |
| 15 import 'package:stack_trace/stack_trace.dart'; |
| 16 |
| 17 /// Uses the inotify subsystem to watch for filesystem events. |
| 18 /// |
| 19 /// Inotify doesn't suport recursively watching subdirectories, nor does |
| 20 /// [Directory.watch] polyfill that functionality. This class polyfills it |
| 21 /// instead. |
| 22 /// |
| 23 /// This class also compensates for the non-inotify-specific issues of |
| 24 /// [Directory.watch] producing multiple events for a single logical action |
| 25 /// (issue 14372) and providing insufficient information about move events |
| 26 /// (issue 14424). |
| 27 class LinuxDirectoryWatcher extends ResubscribableDirectoryWatcher { |
| 28 LinuxDirectoryWatcher(String directory) |
| 29 : super(directory, () => new _LinuxDirectoryWatcher(directory)); |
| 30 } |
| 31 |
| 32 class _LinuxDirectoryWatcher implements ManuallyClosedDirectoryWatcher { |
| 33 final String directory; |
| 34 |
| 35 Stream<WatchEvent> get events => _eventsController.stream; |
| 36 final _eventsController = new StreamController<WatchEvent>.broadcast(); |
| 37 |
| 38 bool get isReady => _readyCompleter.isCompleted; |
| 39 |
| 40 Future get ready => _readyCompleter.future; |
| 41 final _readyCompleter = new Completer(); |
| 42 |
| 43 /// The last known state for each entry in this directory. |
| 44 /// |
| 45 /// The keys in this map are the paths to the directory entries; the values |
| 46 /// are [_EntryState]s indicating whether the entries are files or |
| 47 /// directories. |
| 48 final _entries = new Map<String, _EntryState>(); |
| 49 |
| 50 /// The watchers for subdirectories of [directory]. |
| 51 final _subWatchers = new Map<String, _LinuxDirectoryWatcher>(); |
| 52 |
| 53 /// A set of all subscriptions that this watcher subscribes to. |
| 54 /// |
| 55 /// These are gathered together so that they may all be canceled when the |
| 56 /// watcher is closed. |
| 57 final _subscriptions = new Set<StreamSubscription>(); |
| 58 |
| 59 _LinuxDirectoryWatcher(String directory) |
| 60 : directory = directory { |
| 61 // Batch the inotify changes together so that we can dedup events. |
| 62 var innerStream = new Directory(directory).watch().transform( |
| 63 new BatchedStreamTransformer<FileSystemEvent>()); |
| 64 _listen(innerStream, _onBatch, |
| 65 onError: _eventsController.addError, |
| 66 onDone: _onDone); |
| 67 |
| 68 _listen(new Directory(directory).list(), (entity) { |
| 69 _entries[entity.path] = new _EntryState(entity is Directory); |
| 70 if (entity is! Directory) return; |
| 71 _watchSubdir(entity.path); |
| 72 }, onError: (error, stackTrace) { |
| 73 _eventsController.addError(error, stackTrace); |
| 74 close(); |
| 75 }, onDone: () { |
| 76 _waitUntilReady().then((_) => _readyCompleter.complete()); |
| 77 }, cancelOnError: true); |
| 78 } |
| 79 |
| 80 /// Returns a [Future] that completes once all the subdirectory watchers are |
| 81 /// fully initialized. |
| 82 Future _waitUntilReady() { |
| 83 return Future.wait(_subWatchers.values.map((watcher) => watcher.ready)) |
| 84 .then((_) { |
| 85 if (_subWatchers.values.every((watcher) => watcher.isReady)) return; |
| 86 return _waitUntilReady(); |
| 87 }); |
| 88 } |
| 89 |
| 90 void close() { |
| 91 for (var subscription in _subscriptions) { |
| 92 subscription.cancel(); |
| 93 } |
| 94 for (var watcher in _subWatchers.values) { |
| 95 watcher.close(); |
| 96 } |
| 97 |
| 98 _subWatchers.clear(); |
| 99 _subscriptions.clear(); |
| 100 _eventsController.close(); |
| 101 } |
| 102 |
| 103 /// Returns all files (not directories) that this watcher knows of are |
| 104 /// recursively in the watched directory. |
| 105 Set<String> get _allFiles { |
| 106 var files = new Set<String>(); |
| 107 _getAllFiles(files); |
| 108 return files; |
| 109 } |
| 110 |
| 111 /// Helper function for [_allFiles]. |
| 112 /// |
| 113 /// Adds all files that this watcher knows of to [files]. |
| 114 void _getAllFiles(Set<String> files) { |
| 115 files.addAll(_entries.keys |
| 116 .where((path) => _entries[path] == _EntryState.FILE).toSet()); |
| 117 for (var watcher in _subWatchers.values) { |
| 118 watcher._getAllFiles(files); |
| 119 } |
| 120 } |
| 121 |
| 122 /// Watch a subdirectory of [directory] for changes. |
| 123 /// |
| 124 /// If the subdirectory was added after [this] began emitting events, its |
| 125 /// contents will be emitted as ADD events. |
| 126 void _watchSubdir(String path) { |
| 127 if (_subWatchers.containsKey(path)) return; |
| 128 var watcher = new _LinuxDirectoryWatcher(path); |
| 129 _subWatchers[path] = watcher; |
| 130 |
| 131 // TODO(nweiz): Catch any errors here that indicate that the directory in |
| 132 // question doesn't exist and silently stop watching it instead of |
| 133 // propagating the errors. |
| 134 _listen(watcher.events, (event) { |
| 135 if (isReady) _eventsController.add(event); |
| 136 }, onError: (error, stackTrace) { |
| 137 _eventsController.addError(error, stackTrace); |
| 138 _eventsController.close(); |
| 139 }, onDone: () { |
| 140 if (_subWatchers[path] == watcher) _subWatchers.remove(path); |
| 141 |
| 142 // It's possible that a directory was removed and recreated very quickly. |
| 143 // If so, make sure we're still watching it. |
| 144 if (new Directory(path).existsSync()) _watchSubdir(path); |
| 145 }); |
| 146 |
| 147 // TODO(nweiz): Right now it's possible for the watcher to emit an event for |
| 148 // a file before the directory list is complete. This could lead to the user |
| 149 // seeing a MODIFY or REMOVE event for a file before they see an ADD event, |
| 150 // which is bad. We should handle that. |
| 151 // |
| 152 // One possibility is to provide a general means (e.g. |
| 153 // `DirectoryWatcher.eventsAndExistingFiles`) to tell a watcher to emit |
| 154 // events for all the files that already exist. This would be useful for |
| 155 // top-level clients such as barback as well, and could be implemented with |
| 156 // a wrapper similar to how listening/canceling works now. |
| 157 |
| 158 // If a directory is added after we're finished with the initial scan, emit |
| 159 // an event for each entry in it. This gives the user consistently gets an |
| 160 // event for every new file. |
| 161 watcher.ready.then((_) { |
| 162 if (!isReady || _eventsController.isClosed) return; |
| 163 _listen(new Directory(path).list(recursive: true), (entry) { |
| 164 if (entry is Directory) return; |
| 165 _eventsController.add(new WatchEvent(ChangeType.ADD, entry.path)); |
| 166 }, onError: (error, stackTrace) { |
| 167 // Ignore an exception caused by the dir not existing. It's fine if it |
| 168 // was added and then quickly removed. |
| 169 if (error is FileSystemException) return; |
| 170 |
| 171 _eventsController.addError(error, stackTrace); |
| 172 _eventsController.close(); |
| 173 }, cancelOnError: true); |
| 174 }); |
| 175 } |
| 176 |
| 177 /// The callback that's run when a batch of changes comes in. |
| 178 void _onBatch(List<FileSystemEvent> batch) { |
| 179 var changedEntries = new Set<String>(); |
| 180 var oldEntries = new Map.from(_entries); |
| 181 |
| 182 // inotify event batches are ordered by occurrence, so we treat them as a |
| 183 // log of what happened to a file. |
| 184 for (var event in batch) { |
| 185 // If the watched directory is deleted or moved, we'll get a deletion |
| 186 // event for it. Ignore it; we handle closing [this] when the underlying |
| 187 // stream is closed. |
| 188 if (event.path == directory) continue; |
| 189 |
| 190 changedEntries.add(event.path); |
| 191 |
| 192 if (event is FileSystemMoveEvent) { |
| 193 changedEntries.add(event.destination); |
| 194 _changeEntryState(event.path, ChangeType.REMOVE, event.isDirectory); |
| 195 _changeEntryState(event.destination, ChangeType.ADD, event.isDirectory); |
| 196 } else { |
| 197 _changeEntryState(event.path, _changeTypeFor(event), event.isDirectory); |
| 198 } |
| 199 } |
| 200 |
| 201 for (var path in changedEntries) { |
| 202 emitEvent(ChangeType type) { |
| 203 if (isReady) _eventsController.add(new WatchEvent(type, path)); |
| 204 } |
| 205 |
| 206 var oldState = oldEntries[path]; |
| 207 var newState = _entries[path]; |
| 208 |
| 209 if (oldState != _EntryState.FILE && newState == _EntryState.FILE) { |
| 210 emitEvent(ChangeType.ADD); |
| 211 } else if (oldState == _EntryState.FILE && newState == _EntryState.FILE) { |
| 212 emitEvent(ChangeType.MODIFY); |
| 213 } else if (oldState == _EntryState.FILE && newState != _EntryState.FILE) { |
| 214 emitEvent(ChangeType.REMOVE); |
| 215 } |
| 216 |
| 217 if (oldState == _EntryState.DIRECTORY) { |
| 218 var watcher = _subWatchers.remove(path); |
| 219 if (watcher == null) return; |
| 220 for (var path in watcher._allFiles) { |
| 221 _eventsController.add(new WatchEvent(ChangeType.REMOVE, path)); |
| 222 } |
| 223 watcher.close(); |
| 224 } |
| 225 |
| 226 if (newState == _EntryState.DIRECTORY) _watchSubdir(path); |
| 227 } |
| 228 } |
| 229 |
| 230 /// Changes the known state of the entry at [path] based on [change] and |
| 231 /// [isDir]. |
| 232 void _changeEntryState(String path, ChangeType change, bool isDir) { |
| 233 if (change == ChangeType.ADD || change == ChangeType.MODIFY) { |
| 234 _entries[path] = new _EntryState(isDir); |
| 235 } else { |
| 236 assert(change == ChangeType.REMOVE); |
| 237 _entries.remove(path); |
| 238 } |
| 239 } |
| 240 |
| 241 /// Determines the [ChangeType] associated with [event]. |
| 242 ChangeType _changeTypeFor(FileSystemEvent event) { |
| 243 if (event is FileSystemDeleteEvent) return ChangeType.REMOVE; |
| 244 if (event is FileSystemCreateEvent) return ChangeType.ADD; |
| 245 |
| 246 assert(event is FileSystemModifyEvent); |
| 247 return ChangeType.MODIFY; |
| 248 } |
| 249 |
| 250 /// Handles the underlying event stream closing, indicating that the directory |
| 251 /// being watched was removed. |
| 252 void _onDone() { |
| 253 // The parent directory often gets a close event before the subdirectories |
| 254 // are done emitting events. We wait for them to finish before we close |
| 255 // [events] so that we can be sure to emit a remove event for every file |
| 256 // that used to exist. |
| 257 Future.wait(_subWatchers.values.map((watcher) { |
| 258 try { |
| 259 return watcher.events.toList(); |
| 260 } on StateError catch (_) { |
| 261 // It's possible that [watcher.events] is closed but the onDone event |
| 262 // hasn't reached us yet. It's fine if so. |
| 263 return new Future.value(); |
| 264 } |
| 265 })).then((_) => close()); |
| 266 } |
| 267 |
| 268 /// Like [Stream.listen], but automatically adds the subscription to |
| 269 /// [_subscriptions] so that it can be canceled when [close] is called. |
| 270 void _listen(Stream stream, void onData(event), {Function onError, |
| 271 void onDone(), bool cancelOnError}) { |
| 272 var subscription; |
| 273 subscription = stream.listen(onData, onError: onError, onDone: () { |
| 274 _subscriptions.remove(subscription); |
| 275 if (onDone != null) onDone(); |
| 276 }, cancelOnError: cancelOnError); |
| 277 _subscriptions.add(subscription); |
| 278 } |
| 279 } |
| 280 |
| 281 /// An enum for the possible states of entries in a watched directory. |
| 282 class _EntryState { |
| 283 final String _name; |
| 284 |
| 285 /// The entry is a file. |
| 286 static const FILE = const _EntryState._("file"); |
| 287 |
| 288 /// The entry is a directory. |
| 289 static const DIRECTORY = const _EntryState._("directory"); |
| 290 |
| 291 const _EntryState._(this._name); |
| 292 |
| 293 /// Returns [DIRECTORY] if [isDir] is true, and [FILE] otherwise. |
| 294 factory _EntryState(bool isDir) => |
| 295 isDir ? _EntryState.DIRECTORY : _EntryState.FILE; |
| 296 |
| 297 String toString() => _name; |
| 298 } |
OLD | NEW |