Chromium Code Reviews
chromiumcodereview-hr@appspot.gserviceaccount.com (chromiumcodereview-hr) | Please choose your nickname with Settings | Help | Chromium Project | Gerrit Changes | Sign out
(744)

Side by Side Diff: pkg/watcher/lib/src/directory_watcher/linux.dart

Issue 46843003: Wrap Directory.watch on linux for the watcher package. (Closed) Base URL: https://dart.googlecode.com/svn/branches/bleeding_edge/dart
Patch Set: code review Created 7 years, 1 month ago
Use n/p to move between diff chunks; N/P to move between comments. Draft comments are only viewable by you.
Jump to:
View unified diff | Download patch | Annotate | Revision Log
OLDNEW
(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 }
OLDNEW
« no previous file with comments | « pkg/watcher/lib/src/directory_watcher.dart ('k') | pkg/watcher/lib/src/directory_watcher/polling.dart » ('j') | no next file with comments »

Powered by Google App Engine
This is Rietveld 408576698