Wrap Directory.watch on linux for the watcher package.
R=rnystrom@google.com
BUG=14428
Review URL: https://codereview.chromium.org//46843003
git-svn-id: https://dart.googlecode.com/svn/branches/bleeding_edge/dart/pkg/watcher@30081 260f80e4-7a28-3924-810f-c04153c831b5
diff --git a/pkgs/watcher/lib/src/directory_watcher.dart b/pkgs/watcher/lib/src/directory_watcher.dart
index 679e227..4484c2b 100644
--- a/pkgs/watcher/lib/src/directory_watcher.dart
+++ b/pkgs/watcher/lib/src/directory_watcher.dart
@@ -7,18 +7,15 @@
import 'dart:async';
import 'dart:io';
-import 'package:crypto/crypto.dart';
-
-import 'async_queue.dart';
-import 'stat.dart';
-import 'utils.dart';
import 'watch_event.dart';
+import 'directory_watcher/linux.dart';
+import 'directory_watcher/polling.dart';
/// Watches the contents of a directory and emits [WatchEvent]s when something
/// in the directory has changed.
-class DirectoryWatcher {
+abstract class DirectoryWatcher {
/// The directory whose contents are being monitored.
- final String directory;
+ String get directory;
/// The broadcast [Stream] of events that have occurred to files in
/// [directory].
@@ -26,10 +23,12 @@
/// Changes will only be monitored while this stream has subscribers. Any
/// file changes that occur during periods when there are no subscribers
/// will not be reported the next time a subscriber is added.
- Stream<WatchEvent> get events => _events.stream;
- StreamController<WatchEvent> _events;
+ Stream<WatchEvent> get events;
- _WatchState _state = _WatchState.UNSUBSCRIBED;
+ /// Whether the watcher is initialized and watching for file changes.
+ ///
+ /// This is true if and only if [ready] is complete.
+ bool get isReady;
/// A [Future] that completes when the watcher is initialized and watching
/// for file changes.
@@ -41,241 +40,20 @@
///
/// If the watcher is already monitoring, this returns an already complete
/// future.
- Future get ready => _ready.future;
- Completer _ready = new Completer();
-
- /// The amount of time the watcher pauses between successive polls of the
- /// directory contents.
- final Duration pollingDelay;
-
- /// The previous status of the files in the directory.
- ///
- /// Used to tell which files have been modified.
- final _statuses = new Map<String, _FileStatus>();
-
- /// The subscription used while [directory] is being listed.
- ///
- /// Will be `null` if a list is not currently happening.
- StreamSubscription<FileSystemEntity> _listSubscription;
-
- /// The queue of files waiting to be processed to see if they have been
- /// modified.
- ///
- /// Processing a file is asynchronous, as is listing the directory, so the
- /// queue exists to let each of those proceed at their own rate. The lister
- /// will enqueue files as quickly as it can. Meanwhile, files are dequeued
- /// and processed sequentially.
- AsyncQueue<String> _filesToProcess;
-
- /// The set of files that have been seen in the current directory listing.
- ///
- /// Used to tell which files have been removed: files that are in [_statuses]
- /// but not in here when a poll completes have been removed.
- final _polledFiles = new Set<String>();
+ Future get ready;
/// Creates a new [DirectoryWatcher] monitoring [directory].
///
- /// If [pollingDelay] is passed, it specifies the amount of time the watcher
- /// will pause between successive polls of the directory contents. Making
- /// this shorter will give more immediate feedback at the expense of doing
- /// more IO and higher CPU usage. Defaults to one second.
- DirectoryWatcher(this.directory, {Duration pollingDelay})
- : pollingDelay = pollingDelay != null ? pollingDelay :
- new Duration(seconds: 1) {
- _events = new StreamController<WatchEvent>.broadcast(
- onListen: _watch, onCancel: _cancel);
-
- _filesToProcess = new AsyncQueue<String>(_processFile,
- onError: _events.addError);
- }
-
- /// Scans to see which files were already present before the watcher was
- /// subscribed to, and then starts watching the directory for changes.
- void _watch() {
- assert(_state == _WatchState.UNSUBSCRIBED);
- _state = _WatchState.SCANNING;
- _poll();
- }
-
- /// Stops watching the directory when there are no more subscribers.
- void _cancel() {
- assert(_state != _WatchState.UNSUBSCRIBED);
- _state = _WatchState.UNSUBSCRIBED;
-
- // If we're in the middle of listing the directory, stop.
- if (_listSubscription != null) _listSubscription.cancel();
-
- // Don't process any remaining files.
- _filesToProcess.clear();
- _polledFiles.clear();
- _statuses.clear();
-
- _ready = new Completer();
- }
-
- /// Scans the contents of the directory once to see which files have been
- /// added, removed, and modified.
- void _poll() {
- _filesToProcess.clear();
- _polledFiles.clear();
-
- endListing() {
- assert(_state != _WatchState.UNSUBSCRIBED);
- _listSubscription = null;
-
- // Null tells the queue consumer that we're done listing.
- _filesToProcess.add(null);
- }
-
- var stream = new Directory(directory).list(recursive: true);
- _listSubscription = stream.listen((entity) {
- assert(_state != _WatchState.UNSUBSCRIBED);
-
- if (entity is! File) return;
- _filesToProcess.add(entity.path);
- }, onError: (error, StackTrace stackTrace) {
- if (!isDirectoryNotFoundException(error)) {
- // It's some unknown error. Pipe it over to the event stream so the
- // user can see it.
- _events.addError(error, stackTrace);
- }
-
- // When an error occurs, we end the listing normally, which has the
- // desired effect of marking all files that were in the directory as
- // being removed.
- endListing();
- }, onDone: endListing, cancelOnError: true);
- }
-
- /// Processes [file] to determine if it has been modified since the last
- /// time it was scanned.
- Future _processFile(String file) {
- assert(_state != _WatchState.UNSUBSCRIBED);
-
- // `null` is the sentinel which means the directory listing is complete.
- if (file == null) return _completePoll();
-
- return getModificationTime(file).then((modified) {
- if (_checkForCancel()) return null;
-
- var lastStatus = _statuses[file];
-
- // If its modification time hasn't changed, assume the file is unchanged.
- if (lastStatus != null && lastStatus.modified == modified) {
- // The file is still here.
- _polledFiles.add(file);
- return null;
- }
-
- return _hashFile(file).then((hash) {
- if (_checkForCancel()) return;
-
- var status = new _FileStatus(modified, hash);
- _statuses[file] = status;
- _polledFiles.add(file);
-
- // Only notify while in the watching state.
- if (_state != _WatchState.WATCHING) return;
-
- // And the file is different.
- var changed = lastStatus == null || !_sameHash(lastStatus.hash, hash);
- if (!changed) return;
-
- var type = lastStatus == null ? ChangeType.ADD : ChangeType.MODIFY;
- _events.add(new WatchEvent(type, file));
- });
- });
- }
-
- /// After the directory listing is complete, this determines which files were
- /// removed and then restarts the next poll.
- Future _completePoll() {
- // Any files that were not seen in the last poll but that we have a
- // status for must have been removed.
- var removedFiles = _statuses.keys.toSet().difference(_polledFiles);
- for (var removed in removedFiles) {
- if (_state == _WatchState.WATCHING) {
- _events.add(new WatchEvent(ChangeType.REMOVE, removed));
- }
- _statuses.remove(removed);
- }
-
- if (_state == _WatchState.SCANNING) {
- _state = _WatchState.WATCHING;
- _ready.complete();
- }
-
- // Wait and then poll again.
- return new Future.delayed(pollingDelay).then((_) {
- if (_checkForCancel()) return;
- _poll();
- });
- }
-
- /// Returns `true` and clears the processing queue if the watcher has been
- /// unsubscribed.
- bool _checkForCancel() {
- if (_state != _WatchState.UNSUBSCRIBED) return false;
-
- // Don't process any more files.
- _filesToProcess.clear();
- return true;
- }
-
- /// Calculates the SHA-1 hash of the file at [path].
- Future<List<int>> _hashFile(String path) {
- return new File(path).readAsBytes().then((bytes) {
- var sha1 = new SHA1();
- sha1.add(bytes);
- return sha1.close();
- });
- }
-
- /// Returns `true` if [a] and [b] are the same hash value, i.e. the same
- /// series of byte values.
- bool _sameHash(List<int> a, List<int> b) {
- // Hashes should always be the same size.
- assert(a.length == b.length);
-
- for (var i = 0; i < a.length; i++) {
- if (a[i] != b[i]) return false;
- }
-
- return true;
- }
-}
-
-/// Enum class for the states that the [DirectoryWatcher] can be in.
-class _WatchState {
- /// There are no subscribers to the watcher's event stream and no watching
- /// is going on.
- static const UNSUBSCRIBED = const _WatchState("unsubscribed");
-
- /// There are subscribers and the watcher is doing an initial scan of the
- /// directory to see which files were already present before watching started.
+ /// If a native directory watcher is available for this platform, this will
+ /// use it. Otherwise, it will fall back to a [PollingDirectoryWatcher].
///
- /// The watcher does not send notifications for changes that occurred while
- /// there were no subscribers, or for files already present before watching.
- /// The initial scan is used to determine what "before watching" state of
- /// the file system was.
- static const SCANNING = const _WatchState("scanning");
-
- /// There are subscribers and the watcher is polling the directory to look
- /// for changes.
- static const WATCHING = const _WatchState("watching");
-
- /// The name of the state.
- final String name;
-
- const _WatchState(this.name);
+ /// If [_pollingDelay] is passed, it specifies the amount of time the watcher
+ /// will pause between successive polls of the directory contents. Making this
+ /// shorter will give more immediate feedback at the expense of doing more IO
+ /// and higher CPU usage. Defaults to one second. Ignored for non-polling
+ /// watchers.
+ factory DirectoryWatcher(String directory, {Duration pollingDelay}) {
+ if (Platform.isLinux) return new LinuxDirectoryWatcher(directory);
+ return new PollingDirectoryWatcher(directory, pollingDelay: pollingDelay);
+ }
}
-
-class _FileStatus {
- /// The last time the file was modified.
- DateTime modified;
-
- /// The SHA-1 hash of the contents of the file.
- List<int> hash;
-
- _FileStatus(this.modified, this.hash);
-}
\ No newline at end of file
diff --git a/pkgs/watcher/lib/src/directory_watcher/linux.dart b/pkgs/watcher/lib/src/directory_watcher/linux.dart
new file mode 100644
index 0000000..9acecf1
--- /dev/null
+++ b/pkgs/watcher/lib/src/directory_watcher/linux.dart
@@ -0,0 +1,298 @@
+// Copyright (c) 2013, the Dart project authors. Please see the AUTHORS file
+// for details. All rights reserved. Use of this source code is governed by a
+// BSD-style license that can be found in the LICENSE file.
+
+library watcher.directory_watcher.linux;
+
+import 'dart:async';
+import 'dart:io';
+
+import '../directory_watcher.dart';
+import '../utils.dart';
+import '../watch_event.dart';
+import 'resubscribable.dart';
+
+import 'package:stack_trace/stack_trace.dart';
+
+/// Uses the inotify subsystem to watch for filesystem events.
+///
+/// Inotify doesn't suport recursively watching subdirectories, nor does
+/// [Directory.watch] polyfill that functionality. This class polyfills it
+/// instead.
+///
+/// This class also compensates for the non-inotify-specific issues of
+/// [Directory.watch] producing multiple events for a single logical action
+/// (issue 14372) and providing insufficient information about move events
+/// (issue 14424).
+class LinuxDirectoryWatcher extends ResubscribableDirectoryWatcher {
+ LinuxDirectoryWatcher(String directory)
+ : super(directory, () => new _LinuxDirectoryWatcher(directory));
+}
+
+class _LinuxDirectoryWatcher implements ManuallyClosedDirectoryWatcher {
+ final String directory;
+
+ Stream<WatchEvent> get events => _eventsController.stream;
+ final _eventsController = new StreamController<WatchEvent>.broadcast();
+
+ bool get isReady => _readyCompleter.isCompleted;
+
+ Future get ready => _readyCompleter.future;
+ final _readyCompleter = new Completer();
+
+ /// The last known state for each entry in this directory.
+ ///
+ /// The keys in this map are the paths to the directory entries; the values
+ /// are [_EntryState]s indicating whether the entries are files or
+ /// directories.
+ final _entries = new Map<String, _EntryState>();
+
+ /// The watchers for subdirectories of [directory].
+ final _subWatchers = new Map<String, _LinuxDirectoryWatcher>();
+
+ /// A set of all subscriptions that this watcher subscribes to.
+ ///
+ /// These are gathered together so that they may all be canceled when the
+ /// watcher is closed.
+ final _subscriptions = new Set<StreamSubscription>();
+
+ _LinuxDirectoryWatcher(String directory)
+ : directory = directory {
+ // Batch the inotify changes together so that we can dedup events.
+ var innerStream = new Directory(directory).watch().transform(
+ new BatchedStreamTransformer<FileSystemEvent>());
+ _listen(innerStream, _onBatch,
+ onError: _eventsController.addError,
+ onDone: _onDone);
+
+ _listen(new Directory(directory).list(), (entity) {
+ _entries[entity.path] = new _EntryState(entity is Directory);
+ if (entity is! Directory) return;
+ _watchSubdir(entity.path);
+ }, onError: (error, stackTrace) {
+ _eventsController.addError(error, stackTrace);
+ close();
+ }, onDone: () {
+ _waitUntilReady().then((_) => _readyCompleter.complete());
+ }, cancelOnError: true);
+ }
+
+ /// Returns a [Future] that completes once all the subdirectory watchers are
+ /// fully initialized.
+ Future _waitUntilReady() {
+ return Future.wait(_subWatchers.values.map((watcher) => watcher.ready))
+ .then((_) {
+ if (_subWatchers.values.every((watcher) => watcher.isReady)) return;
+ return _waitUntilReady();
+ });
+ }
+
+ void close() {
+ for (var subscription in _subscriptions) {
+ subscription.cancel();
+ }
+ for (var watcher in _subWatchers.values) {
+ watcher.close();
+ }
+
+ _subWatchers.clear();
+ _subscriptions.clear();
+ _eventsController.close();
+ }
+
+ /// Returns all files (not directories) that this watcher knows of are
+ /// recursively in the watched directory.
+ Set<String> get _allFiles {
+ var files = new Set<String>();
+ _getAllFiles(files);
+ return files;
+ }
+
+ /// Helper function for [_allFiles].
+ ///
+ /// Adds all files that this watcher knows of to [files].
+ void _getAllFiles(Set<String> files) {
+ files.addAll(_entries.keys
+ .where((path) => _entries[path] == _EntryState.FILE).toSet());
+ for (var watcher in _subWatchers.values) {
+ watcher._getAllFiles(files);
+ }
+ }
+
+ /// Watch a subdirectory of [directory] for changes.
+ ///
+ /// If the subdirectory was added after [this] began emitting events, its
+ /// contents will be emitted as ADD events.
+ void _watchSubdir(String path) {
+ if (_subWatchers.containsKey(path)) return;
+ var watcher = new _LinuxDirectoryWatcher(path);
+ _subWatchers[path] = watcher;
+
+ // TODO(nweiz): Catch any errors here that indicate that the directory in
+ // question doesn't exist and silently stop watching it instead of
+ // propagating the errors.
+ _listen(watcher.events, (event) {
+ if (isReady) _eventsController.add(event);
+ }, onError: (error, stackTrace) {
+ _eventsController.addError(error, stackTrace);
+ _eventsController.close();
+ }, onDone: () {
+ if (_subWatchers[path] == watcher) _subWatchers.remove(path);
+
+ // It's possible that a directory was removed and recreated very quickly.
+ // If so, make sure we're still watching it.
+ if (new Directory(path).existsSync()) _watchSubdir(path);
+ });
+
+ // TODO(nweiz): Right now it's possible for the watcher to emit an event for
+ // a file before the directory list is complete. This could lead to the user
+ // seeing a MODIFY or REMOVE event for a file before they see an ADD event,
+ // which is bad. We should handle that.
+ //
+ // One possibility is to provide a general means (e.g.
+ // `DirectoryWatcher.eventsAndExistingFiles`) to tell a watcher to emit
+ // events for all the files that already exist. This would be useful for
+ // top-level clients such as barback as well, and could be implemented with
+ // a wrapper similar to how listening/canceling works now.
+
+ // If a directory is added after we're finished with the initial scan, emit
+ // an event for each entry in it. This gives the user consistently gets an
+ // event for every new file.
+ watcher.ready.then((_) {
+ if (!isReady || _eventsController.isClosed) return;
+ _listen(new Directory(path).list(recursive: true), (entry) {
+ if (entry is Directory) return;
+ _eventsController.add(new WatchEvent(ChangeType.ADD, entry.path));
+ }, onError: (error, stackTrace) {
+ // Ignore an exception caused by the dir not existing. It's fine if it
+ // was added and then quickly removed.
+ if (error is FileSystemException) return;
+
+ _eventsController.addError(error, stackTrace);
+ _eventsController.close();
+ }, cancelOnError: true);
+ });
+ }
+
+ /// The callback that's run when a batch of changes comes in.
+ void _onBatch(List<FileSystemEvent> batch) {
+ var changedEntries = new Set<String>();
+ var oldEntries = new Map.from(_entries);
+
+ // inotify event batches are ordered by occurrence, so we treat them as a
+ // log of what happened to a file.
+ for (var event in batch) {
+ // If the watched directory is deleted or moved, we'll get a deletion
+ // event for it. Ignore it; we handle closing [this] when the underlying
+ // stream is closed.
+ if (event.path == directory) continue;
+
+ changedEntries.add(event.path);
+
+ if (event is FileSystemMoveEvent) {
+ changedEntries.add(event.destination);
+ _changeEntryState(event.path, ChangeType.REMOVE, event.isDirectory);
+ _changeEntryState(event.destination, ChangeType.ADD, event.isDirectory);
+ } else {
+ _changeEntryState(event.path, _changeTypeFor(event), event.isDirectory);
+ }
+ }
+
+ for (var path in changedEntries) {
+ emitEvent(ChangeType type) {
+ if (isReady) _eventsController.add(new WatchEvent(type, path));
+ }
+
+ var oldState = oldEntries[path];
+ var newState = _entries[path];
+
+ if (oldState != _EntryState.FILE && newState == _EntryState.FILE) {
+ emitEvent(ChangeType.ADD);
+ } else if (oldState == _EntryState.FILE && newState == _EntryState.FILE) {
+ emitEvent(ChangeType.MODIFY);
+ } else if (oldState == _EntryState.FILE && newState != _EntryState.FILE) {
+ emitEvent(ChangeType.REMOVE);
+ }
+
+ if (oldState == _EntryState.DIRECTORY) {
+ var watcher = _subWatchers.remove(path);
+ if (watcher == null) return;
+ for (var path in watcher._allFiles) {
+ _eventsController.add(new WatchEvent(ChangeType.REMOVE, path));
+ }
+ watcher.close();
+ }
+
+ if (newState == _EntryState.DIRECTORY) _watchSubdir(path);
+ }
+ }
+
+ /// Changes the known state of the entry at [path] based on [change] and
+ /// [isDir].
+ void _changeEntryState(String path, ChangeType change, bool isDir) {
+ if (change == ChangeType.ADD || change == ChangeType.MODIFY) {
+ _entries[path] = new _EntryState(isDir);
+ } else {
+ assert(change == ChangeType.REMOVE);
+ _entries.remove(path);
+ }
+ }
+
+ /// Determines the [ChangeType] associated with [event].
+ ChangeType _changeTypeFor(FileSystemEvent event) {
+ if (event is FileSystemDeleteEvent) return ChangeType.REMOVE;
+ if (event is FileSystemCreateEvent) return ChangeType.ADD;
+
+ assert(event is FileSystemModifyEvent);
+ return ChangeType.MODIFY;
+ }
+
+ /// Handles the underlying event stream closing, indicating that the directory
+ /// being watched was removed.
+ void _onDone() {
+ // The parent directory often gets a close event before the subdirectories
+ // are done emitting events. We wait for them to finish before we close
+ // [events] so that we can be sure to emit a remove event for every file
+ // that used to exist.
+ Future.wait(_subWatchers.values.map((watcher) {
+ try {
+ return watcher.events.toList();
+ } on StateError catch (_) {
+ // It's possible that [watcher.events] is closed but the onDone event
+ // hasn't reached us yet. It's fine if so.
+ return new Future.value();
+ }
+ })).then((_) => close());
+ }
+
+ /// Like [Stream.listen], but automatically adds the subscription to
+ /// [_subscriptions] so that it can be canceled when [close] is called.
+ void _listen(Stream stream, void onData(event), {Function onError,
+ void onDone(), bool cancelOnError}) {
+ var subscription;
+ subscription = stream.listen(onData, onError: onError, onDone: () {
+ _subscriptions.remove(subscription);
+ if (onDone != null) onDone();
+ }, cancelOnError: cancelOnError);
+ _subscriptions.add(subscription);
+ }
+}
+
+/// An enum for the possible states of entries in a watched directory.
+class _EntryState {
+ final String _name;
+
+ /// The entry is a file.
+ static const FILE = const _EntryState._("file");
+
+ /// The entry is a directory.
+ static const DIRECTORY = const _EntryState._("directory");
+
+ const _EntryState._(this._name);
+
+ /// Returns [DIRECTORY] if [isDir] is true, and [FILE] otherwise.
+ factory _EntryState(bool isDir) =>
+ isDir ? _EntryState.DIRECTORY : _EntryState.FILE;
+
+ String toString() => _name;
+}
diff --git a/pkgs/watcher/lib/src/directory_watcher/polling.dart b/pkgs/watcher/lib/src/directory_watcher/polling.dart
new file mode 100644
index 0000000..91ca005
--- /dev/null
+++ b/pkgs/watcher/lib/src/directory_watcher/polling.dart
@@ -0,0 +1,217 @@
+// Copyright (c) 2013, the Dart project authors. Please see the AUTHORS file
+// for details. All rights reserved. Use of this source code is governed by a
+// BSD-style license that can be found in the LICENSE file.
+
+library watcher.directory_watcher.polling;
+
+import 'dart:async';
+import 'dart:io';
+
+import 'package:crypto/crypto.dart';
+
+import '../async_queue.dart';
+import '../directory_watcher.dart';
+import '../stat.dart';
+import '../utils.dart';
+import '../watch_event.dart';
+import 'resubscribable.dart';
+
+/// Periodically polls a directory for changes.
+class PollingDirectoryWatcher extends ResubscribableDirectoryWatcher {
+ /// Creates a new polling watcher monitoring [directory].
+ ///
+ /// If [_pollingDelay] is passed, it specifies the amount of time the watcher
+ /// will pause between successive polls of the directory contents. Making this
+ /// shorter will give more immediate feedback at the expense of doing more IO
+ /// and higher CPU usage. Defaults to one second.
+ PollingDirectoryWatcher(String directory, {Duration pollingDelay})
+ : super(directory, () {
+ return new _PollingDirectoryWatcher(directory,
+ pollingDelay != null ? pollingDelay : new Duration(seconds: 1));
+ });
+}
+
+class _PollingDirectoryWatcher implements ManuallyClosedDirectoryWatcher {
+ final String directory;
+
+ Stream<WatchEvent> get events => _events.stream;
+ final _events = new StreamController<WatchEvent>.broadcast();
+
+ bool get isReady => _ready.isCompleted;
+
+ Future get ready => _ready.future;
+ final _ready = new Completer();
+
+ /// The amount of time the watcher pauses between successive polls of the
+ /// directory contents.
+ final Duration _pollingDelay;
+
+ /// The previous status of the files in the directory.
+ ///
+ /// Used to tell which files have been modified.
+ final _statuses = new Map<String, _FileStatus>();
+
+ /// The subscription used while [directory] is being listed.
+ ///
+ /// Will be `null` if a list is not currently happening.
+ StreamSubscription<FileSystemEntity> _listSubscription;
+
+ /// The queue of files waiting to be processed to see if they have been
+ /// modified.
+ ///
+ /// Processing a file is asynchronous, as is listing the directory, so the
+ /// queue exists to let each of those proceed at their own rate. The lister
+ /// will enqueue files as quickly as it can. Meanwhile, files are dequeued
+ /// and processed sequentially.
+ AsyncQueue<String> _filesToProcess;
+
+ /// The set of files that have been seen in the current directory listing.
+ ///
+ /// Used to tell which files have been removed: files that are in [_statuses]
+ /// but not in here when a poll completes have been removed.
+ final _polledFiles = new Set<String>();
+
+ _PollingDirectoryWatcher(this.directory, this._pollingDelay) {
+ _filesToProcess = new AsyncQueue<String>(_processFile,
+ onError: _events.addError);
+
+ _poll();
+ }
+
+ void close() {
+ _events.close();
+
+ // If we're in the middle of listing the directory, stop.
+ if (_listSubscription != null) _listSubscription.cancel();
+
+ // Don't process any remaining files.
+ _filesToProcess.clear();
+ _polledFiles.clear();
+ _statuses.clear();
+ }
+
+ /// Scans the contents of the directory once to see which files have been
+ /// added, removed, and modified.
+ void _poll() {
+ _filesToProcess.clear();
+ _polledFiles.clear();
+
+ endListing() {
+ assert(!_events.isClosed);
+ _listSubscription = null;
+
+ // Null tells the queue consumer that we're done listing.
+ _filesToProcess.add(null);
+ }
+
+ var stream = new Directory(directory).list(recursive: true);
+ _listSubscription = stream.listen((entity) {
+ assert(!_events.isClosed);
+
+ if (entity is! File) return;
+ _filesToProcess.add(entity.path);
+ }, onError: (error, stackTrace) {
+ if (!isDirectoryNotFoundException(error)) {
+ // It's some unknown error. Pipe it over to the event stream so the
+ // user can see it.
+ _events.addError(error, stackTrace);
+ }
+
+ // When an error occurs, we end the listing normally, which has the
+ // desired effect of marking all files that were in the directory as
+ // being removed.
+ endListing();
+ }, onDone: endListing, cancelOnError: true);
+ }
+
+ /// Processes [file] to determine if it has been modified since the last
+ /// time it was scanned.
+ Future _processFile(String file) {
+ // `null` is the sentinel which means the directory listing is complete.
+ if (file == null) return _completePoll();
+
+ return getModificationTime(file).then((modified) {
+ if (_events.isClosed) return null;
+
+ var lastStatus = _statuses[file];
+
+ // If its modification time hasn't changed, assume the file is unchanged.
+ if (lastStatus != null && lastStatus.modified == modified) {
+ // The file is still here.
+ _polledFiles.add(file);
+ return null;
+ }
+
+ return _hashFile(file).then((hash) {
+ if (_events.isClosed) return;
+
+ var status = new _FileStatus(modified, hash);
+ _statuses[file] = status;
+ _polledFiles.add(file);
+
+ // Only notify if we're ready to emit events.
+ if (!isReady) return;
+
+ // And the file is different.
+ var changed = lastStatus == null || !_sameHash(lastStatus.hash, hash);
+ if (!changed) return;
+
+ var type = lastStatus == null ? ChangeType.ADD : ChangeType.MODIFY;
+ _events.add(new WatchEvent(type, file));
+ });
+ });
+ }
+
+ /// After the directory listing is complete, this determines which files were
+ /// removed and then restarts the next poll.
+ Future _completePoll() {
+ // Any files that were not seen in the last poll but that we have a
+ // status for must have been removed.
+ var removedFiles = _statuses.keys.toSet().difference(_polledFiles);
+ for (var removed in removedFiles) {
+ if (isReady) _events.add(new WatchEvent(ChangeType.REMOVE, removed));
+ _statuses.remove(removed);
+ }
+
+ if (!isReady) _ready.complete();
+
+ // Wait and then poll again.
+ return new Future.delayed(_pollingDelay).then((_) {
+ if (_events.isClosed) return;
+ _poll();
+ });
+ }
+
+ /// Calculates the SHA-1 hash of the file at [path].
+ Future<List<int>> _hashFile(String path) {
+ return new File(path).readAsBytes().then((bytes) {
+ var sha1 = new SHA1();
+ sha1.add(bytes);
+ return sha1.close();
+ });
+ }
+
+ /// Returns `true` if [a] and [b] are the same hash value, i.e. the same
+ /// series of byte values.
+ bool _sameHash(List<int> a, List<int> b) {
+ // Hashes should always be the same size.
+ assert(a.length == b.length);
+
+ for (var i = 0; i < a.length; i++) {
+ if (a[i] != b[i]) return false;
+ }
+
+ return true;
+ }
+}
+
+class _FileStatus {
+ /// The last time the file was modified.
+ DateTime modified;
+
+ /// The SHA-1 hash of the contents of the file.
+ List<int> hash;
+
+ _FileStatus(this.modified, this.hash);
+}
+
diff --git a/pkgs/watcher/lib/src/directory_watcher/resubscribable.dart b/pkgs/watcher/lib/src/directory_watcher/resubscribable.dart
new file mode 100644
index 0000000..daa813a
--- /dev/null
+++ b/pkgs/watcher/lib/src/directory_watcher/resubscribable.dart
@@ -0,0 +1,79 @@
+// Copyright (c) 2013, the Dart project authors. Please see the AUTHORS file
+// for details. All rights reserved. Use of this source code is governed by a
+// BSD-style license that can be found in the LICENSE file.
+
+library watcher.directory_watcher.resubscribable;
+
+import 'dart:async';
+import 'dart:io';
+
+import '../directory_watcher.dart';
+import '../utils.dart';
+import '../watch_event.dart';
+
+typedef ManuallyClosedDirectoryWatcher WatcherFactory();
+
+/// A wrapper for [ManuallyClosedDirectoryWatcher] that encapsulates support for
+/// closing the watcher when it has no subscribers and re-opening it when it's
+/// re-subscribed.
+///
+/// It's simpler to implement watchers without worrying about this behavior.
+/// This class wraps a watcher class which can be written with the simplifying
+/// assumption that it can continue emitting events until an explicit `close`
+/// method is called, at which point it will cease emitting events entirely. The
+/// [ManuallyClosedDirectoryWatcher] interface is used for these watchers.
+///
+/// This would be more cleanly implemented as a function that takes a class and
+/// emits a new class, but Dart doesn't support that sort of thing. Instead it
+/// takes a factory function that produces instances of the inner class.
+abstract class ResubscribableDirectoryWatcher implements DirectoryWatcher {
+ /// The factory function that produces instances of the inner class.
+ final WatcherFactory _factory;
+
+ final String directory;
+
+ Stream<WatchEvent> get events => _eventsController.stream;
+ StreamController<WatchEvent> _eventsController;
+
+ bool get isReady => _readyCompleter.isCompleted;
+
+ Future get ready => _readyCompleter.future;
+ var _readyCompleter = new Completer();
+
+ /// Creates a new [ResubscribableDirectoryWatcher] wrapping the watchers
+ /// emitted by [_factory].
+ ResubscribableDirectoryWatcher(this.directory, this._factory) {
+ var watcher;
+ var subscription;
+
+ _eventsController = new StreamController<WatchEvent>.broadcast(
+ onListen: () {
+ watcher = _factory();
+ subscription = watcher.events.listen(_eventsController.add,
+ onError: _eventsController.addError,
+ onDone: _eventsController.close);
+
+ // It's important that we complete the value of [_readyCompleter] at the
+ // time [onListen] is called, as opposed to the value when [watcher.ready]
+ // fires. A new completer may be created by that time.
+ watcher.ready.then(_readyCompleter.complete);
+ }, onCancel: () {
+ // Cancel the subscription before closing the watcher so that the
+ // watcher's `onDone` event doesn't close [events].
+ subscription.cancel();
+ watcher.close();
+ _readyCompleter = new Completer();
+ }, sync: true);
+ }
+}
+
+/// An interface for watchers with an explicit, manual [close] method.
+///
+/// See [ResubscribableDirectoryWatcher].
+abstract class ManuallyClosedDirectoryWatcher implements DirectoryWatcher {
+ /// Closes the watcher.
+ ///
+ /// Subclasses should close their [events] stream and release any internal
+ /// resources.
+ void close();
+}
diff --git a/pkgs/watcher/lib/src/utils.dart b/pkgs/watcher/lib/src/utils.dart
index 3d00c08..e4e4457 100644
--- a/pkgs/watcher/lib/src/utils.dart
+++ b/pkgs/watcher/lib/src/utils.dart
@@ -4,7 +4,9 @@
library watcher.utils;
+import 'dart:async';
import 'dart:io';
+import 'dart:collection';
/// Returns `true` if [error] is a [FileSystemException] for a missing
/// directory.
@@ -15,3 +17,57 @@
var notFoundCode = Platform.operatingSystem == "windows" ? 3 : 2;
return error.osError.errorCode == notFoundCode;
}
+
+/// Returns a buffered stream that will emit the same values as the stream
+/// returned by [future] once [future] completes.
+///
+/// If [future] completes to an error, the return value will emit that error and
+/// then close.
+Stream futureStream(Future<Stream> future) {
+ var controller = new StreamController(sync: true);
+ future.then((stream) {
+ stream.listen(
+ controller.add,
+ onError: controller.addError,
+ onDone: controller.close);
+ }).catchError((e, stackTrace) {
+ controller.addError(e, stackTrace);
+ controller.close();
+ });
+ return controller.stream;
+}
+
+/// Like [new Future], but avoids around issue 11911 by using [new Future.value]
+/// under the covers.
+Future newFuture(callback()) => new Future.value().then((_) => callback());
+
+/// A stream transformer that batches all events that are sent at the same time.
+///
+/// When multiple events are synchronously added to a stream controller, the
+/// [StreamController] implementation uses [scheduleMicrotask] to schedule the
+/// asynchronous firing of each event. In order to recreate the synchronous
+/// batches, this collates all the events that are received in "nearby"
+/// microtasks.
+class BatchedStreamTransformer<T> implements StreamTransformer<T, List<T>> {
+ Stream<List<T>> bind(Stream<T> input) {
+ var batch = new Queue();
+ return new StreamTransformer<T, List<T>>.fromHandlers(
+ handleData: (event, sink) {
+ batch.add(event);
+
+ // [Timer.run] schedules an event that runs after any microtasks that have
+ // been scheduled.
+ Timer.run(() {
+ if (batch.isEmpty) return;
+ sink.add(batch.toList());
+ batch.clear();
+ });
+ }, handleDone: (sink) {
+ if (batch.isNotEmpty) {
+ sink.add(batch.toList());
+ batch.clear();
+ }
+ sink.close();
+ }).bind(input);
+ }
+}
diff --git a/pkgs/watcher/lib/watcher.dart b/pkgs/watcher/lib/watcher.dart
index c4824b8..88531f2 100644
--- a/pkgs/watcher/lib/watcher.dart
+++ b/pkgs/watcher/lib/watcher.dart
@@ -6,3 +6,4 @@
export 'src/watch_event.dart';
export 'src/directory_watcher.dart';
+export 'src/directory_watcher/polling.dart';
diff --git a/pkgs/watcher/test/directory_watcher/linux_test.dart b/pkgs/watcher/test/directory_watcher/linux_test.dart
new file mode 100644
index 0000000..ba69569
--- /dev/null
+++ b/pkgs/watcher/test/directory_watcher/linux_test.dart
@@ -0,0 +1,62 @@
+// Copyright (c) 2013, the Dart project authors. Please see the AUTHORS file
+// for details. All rights reserved. Use of this source code is governed by a
+// BSD-style license that can be found in the LICENSE file.
+
+import 'package:scheduled_test/scheduled_test.dart';
+import 'package:watcher/src/directory_watcher/linux.dart';
+import 'package:watcher/watcher.dart';
+
+import 'shared.dart';
+import '../utils.dart';
+
+main() {
+ initConfig();
+
+ watcherFactory = (dir) => new LinuxDirectoryWatcher(dir);
+
+ setUp(() {
+ // Increase the timeout because closing a [Directory.watch] stream blocks
+ // the main isolate for a very long time on Goobuntu, as of kernel
+ // 3.2.5-gg1336 (see issue 14606).
+ currentSchedule.timeout *= 3;
+
+ createSandbox();
+ });
+
+ sharedTests();
+
+ test('DirectoryWatcher creates a LinuxDirectoryWatcher on Linux', () {
+ expect(new DirectoryWatcher('.'),
+ new isInstanceOf<LinuxDirectoryWatcher>());
+ });
+
+ test('notifies even if the file contents are unchanged', () {
+ writeFile("a.txt", contents: "same");
+ writeFile("b.txt", contents: "before");
+ startWatcher();
+ writeFile("a.txt", contents: "same");
+ writeFile("b.txt", contents: "after");
+ expectModifyEvent("a.txt");
+ expectModifyEvent("b.txt");
+ });
+
+ test('emits events for many nested files moved out then immediately back in',
+ () {
+ withPermutations((i, j, k) =>
+ writeFile("dir/sub/sub-$i/sub-$j/file-$k.txt"));
+ startWatcher(dir: "dir");
+
+ renameDir("dir/sub", "sub");
+ renameDir("sub", "dir/sub");
+
+ inAnyOrder(() {
+ withPermutations((i, j, k) =>
+ expectRemoveEvent("dir/sub/sub-$i/sub-$j/file-$k.txt"));
+ });
+
+ inAnyOrder(() {
+ withPermutations((i, j, k) =>
+ expectAddEvent("dir/sub/sub-$i/sub-$j/file-$k.txt"));
+ });
+ });
+}
diff --git a/pkgs/watcher/test/directory_watcher/polling_test.dart b/pkgs/watcher/test/directory_watcher/polling_test.dart
new file mode 100644
index 0000000..02ed5d2
--- /dev/null
+++ b/pkgs/watcher/test/directory_watcher/polling_test.dart
@@ -0,0 +1,39 @@
+// Copyright (c) 2013, the Dart project authors. Please see the AUTHORS file
+// for details. All rights reserved. Use of this source code is governed by a
+// BSD-style license that can be found in the LICENSE file.
+
+import 'package:scheduled_test/scheduled_test.dart';
+import 'package:watcher/watcher.dart';
+
+import 'shared.dart';
+import '../utils.dart';
+
+main() {
+ initConfig();
+
+ // Use a short delay to make the tests run quickly.
+ watcherFactory = (dir) => new PollingDirectoryWatcher(dir,
+ pollingDelay: new Duration(milliseconds: 100));
+
+ setUp(createSandbox);
+
+ sharedTests();
+
+ test('does not notify if the file contents are unchanged', () {
+ writeFile("a.txt", contents: "same");
+ writeFile("b.txt", contents: "before");
+ startWatcher();
+ writeFile("a.txt", contents: "same");
+ writeFile("b.txt", contents: "after");
+ expectModifyEvent("b.txt");
+ });
+
+ test('does not notify if the modification time did not change', () {
+ writeFile("a.txt", contents: "before");
+ writeFile("b.txt", contents: "before");
+ startWatcher();
+ writeFile("a.txt", contents: "after", updateModified: false);
+ writeFile("b.txt", contents: "after");
+ expectModifyEvent("b.txt");
+ });
+}
diff --git a/pkgs/watcher/test/directory_watcher/shared.dart b/pkgs/watcher/test/directory_watcher/shared.dart
new file mode 100644
index 0000000..eed6069
--- /dev/null
+++ b/pkgs/watcher/test/directory_watcher/shared.dart
@@ -0,0 +1,222 @@
+// Copyright (c) 2012, the Dart project authors. Please see the AUTHORS file
+// for details. All rights reserved. Use of this source code is governed by a
+// BSD-style license that can be found in the LICENSE file.
+
+import 'package:scheduled_test/scheduled_test.dart';
+
+import '../utils.dart';
+
+sharedTests() {
+ test('does not notify for files that already exist when started', () {
+ // Make some pre-existing files.
+ writeFile("a.txt");
+ writeFile("b.txt");
+
+ startWatcher();
+
+ // Change one after the watcher is running.
+ writeFile("b.txt", contents: "modified");
+
+ // We should get a modify event for the changed file, but no add events
+ // for them before this.
+ expectModifyEvent("b.txt");
+ });
+
+ test('notifies when a file is added', () {
+ startWatcher();
+ writeFile("file.txt");
+ expectAddEvent("file.txt");
+ });
+
+ test('notifies when a file is modified', () {
+ writeFile("file.txt");
+ startWatcher();
+ writeFile("file.txt", contents: "modified");
+ expectModifyEvent("file.txt");
+ });
+
+ test('notifies when a file is removed', () {
+ writeFile("file.txt");
+ startWatcher();
+ deleteFile("file.txt");
+ expectRemoveEvent("file.txt");
+ });
+
+ test('notifies when a file is modified multiple times', () {
+ writeFile("file.txt");
+ startWatcher();
+ writeFile("file.txt", contents: "modified");
+ expectModifyEvent("file.txt");
+ writeFile("file.txt", contents: "modified again");
+ expectModifyEvent("file.txt");
+ });
+
+ test('when the watched directory is deleted, removes all files', () {
+ writeFile("dir/a.txt");
+ writeFile("dir/b.txt");
+
+ startWatcher(dir: "dir");
+
+ deleteDir("dir");
+ inAnyOrder(() {
+ expectRemoveEvent("dir/a.txt");
+ expectRemoveEvent("dir/b.txt");
+ });
+ });
+
+ group("moves", () {
+ test('notifies when a file is moved within the watched directory', () {
+ writeFile("old.txt");
+ startWatcher();
+ renameFile("old.txt", "new.txt");
+
+ inAnyOrder(() {
+ expectAddEvent("new.txt");
+ expectRemoveEvent("old.txt");
+ });
+ });
+
+ test('notifies when a file is moved from outside the watched directory',
+ () {
+ writeFile("old.txt");
+ createDir("dir");
+ startWatcher(dir: "dir");
+
+ renameFile("old.txt", "dir/new.txt");
+ expectAddEvent("dir/new.txt");
+ });
+
+ test('notifies when a file is moved outside the watched directory', () {
+ writeFile("dir/old.txt");
+ startWatcher(dir: "dir");
+
+ renameFile("dir/old.txt", "new.txt");
+ expectRemoveEvent("dir/old.txt");
+ });
+ });
+
+ group("clustered changes", () {
+ test("doesn't notify when a file is created and then immediately removed",
+ () {
+ startWatcher();
+ writeFile("file.txt");
+ deleteFile("file.txt");
+
+ // [startWatcher] will assert that no events were fired.
+ });
+
+ test("reports a modification when a file is deleted and then immediately "
+ "recreated", () {
+ writeFile("file.txt");
+ startWatcher();
+
+ deleteFile("file.txt");
+ writeFile("file.txt", contents: "re-created");
+ expectModifyEvent("file.txt");
+ });
+
+ test("reports a modification when a file is moved and then immediately "
+ "recreated", () {
+ writeFile("old.txt");
+ startWatcher();
+
+ renameFile("old.txt", "new.txt");
+ writeFile("old.txt", contents: "re-created");
+ inAnyOrder(() {
+ expectModifyEvent("old.txt");
+ expectAddEvent("new.txt");
+ });
+ });
+
+ test("reports a removal when a file is modified and then immediately "
+ "removed", () {
+ writeFile("file.txt");
+ startWatcher();
+
+ writeFile("file.txt", contents: "modified");
+ deleteFile("file.txt");
+ expectRemoveEvent("file.txt");
+ });
+
+ test("reports an add when a file is added and then immediately modified",
+ () {
+ startWatcher();
+
+ writeFile("file.txt");
+ writeFile("file.txt", contents: "modified");
+ expectAddEvent("file.txt");
+ });
+ });
+
+ group("subdirectories", () {
+ test('watches files in subdirectories', () {
+ startWatcher();
+ writeFile("a/b/c/d/file.txt");
+ expectAddEvent("a/b/c/d/file.txt");
+ });
+
+ test('notifies when a subdirectory is moved within the watched directory '
+ 'and then its contents are modified', () {
+ writeFile("old/file.txt");
+ startWatcher();
+
+ renameDir("old", "new");
+ inAnyOrder(() {
+ expectRemoveEvent("old/file.txt");
+ expectAddEvent("new/file.txt");
+ });
+
+ writeFile("new/file.txt", contents: "modified");
+ expectModifyEvent("new/file.txt");
+ });
+
+ test('emits events for many nested files added at once', () {
+ withPermutations((i, j, k) =>
+ writeFile("sub/sub-$i/sub-$j/file-$k.txt"));
+
+ createDir("dir");
+ startWatcher(dir: "dir");
+ renameDir("sub", "dir/sub");
+
+ inAnyOrder(() {
+ withPermutations((i, j, k) =>
+ expectAddEvent("dir/sub/sub-$i/sub-$j/file-$k.txt"));
+ });
+ });
+
+ test('emits events for many nested files removed at once', () {
+ withPermutations((i, j, k) =>
+ writeFile("dir/sub/sub-$i/sub-$j/file-$k.txt"));
+
+ createDir("dir");
+ startWatcher(dir: "dir");
+
+ // Rename the directory rather than deleting it because native watchers
+ // report a rename as a single DELETE event for the directory, whereas
+ // they report recursive deletion with DELETE events for every file in the
+ // directory.
+ renameDir("dir/sub", "sub");
+
+ inAnyOrder(() {
+ withPermutations((i, j, k) =>
+ expectRemoveEvent("dir/sub/sub-$i/sub-$j/file-$k.txt"));
+ });
+ });
+
+ test('emits events for many nested files moved at once', () {
+ withPermutations((i, j, k) =>
+ writeFile("dir/old/sub-$i/sub-$j/file-$k.txt"));
+
+ createDir("dir");
+ startWatcher(dir: "dir");
+ renameDir("dir/old", "dir/new");
+
+ inAnyOrder(() {
+ withPermutations((i, j, k) {
+ expectRemoveEvent("dir/old/sub-$i/sub-$j/file-$k.txt");
+ expectAddEvent("dir/new/sub-$i/sub-$j/file-$k.txt");
+ });
+ });
+ });
+ });
+}
diff --git a/pkgs/watcher/test/directory_watcher_test.dart b/pkgs/watcher/test/directory_watcher_test.dart
deleted file mode 100644
index 841dd08..0000000
--- a/pkgs/watcher/test/directory_watcher_test.dart
+++ /dev/null
@@ -1,108 +0,0 @@
-// Copyright (c) 2012, the Dart project authors. Please see the AUTHORS file
-// for details. All rights reserved. Use of this source code is governed by a
-// BSD-style license that can be found in the LICENSE file.
-
-import 'package:scheduled_test/scheduled_test.dart';
-
-import 'utils.dart';
-
-main() {
- initConfig();
-
- setUp(createSandbox);
-
- test('does not notify for files that already exist when started', () {
- // Make some pre-existing files.
- writeFile("a.txt");
- writeFile("b.txt");
-
- createWatcher();
-
- // Change one after the watcher is running.
- writeFile("b.txt", contents: "modified");
-
- // We should get a modify event for the changed file, but no add events
- // for them before this.
- expectModifyEvent("b.txt");
- });
-
- test('notifies when a file is added', () {
- createWatcher();
- writeFile("file.txt");
- expectAddEvent("file.txt");
- });
-
- test('notifies when a file is modified', () {
- writeFile("file.txt");
- createWatcher();
- writeFile("file.txt", contents: "modified");
- expectModifyEvent("file.txt");
- });
-
- test('notifies when a file is removed', () {
- writeFile("file.txt");
- createWatcher();
- deleteFile("file.txt");
- expectRemoveEvent("file.txt");
- });
-
- test('notifies when a file is moved', () {
- writeFile("old.txt");
- createWatcher();
- renameFile("old.txt", "new.txt");
- expectAddEvent("new.txt");
- expectRemoveEvent("old.txt");
- });
-
- test('notifies when a file is modified multiple times', () {
- writeFile("file.txt");
- createWatcher();
- writeFile("file.txt", contents: "modified");
- expectModifyEvent("file.txt");
- writeFile("file.txt", contents: "modified again");
- expectModifyEvent("file.txt");
- });
-
- test('does not notify if the file contents are unchanged', () {
- writeFile("a.txt", contents: "same");
- writeFile("b.txt", contents: "before");
- createWatcher();
- writeFile("a.txt", contents: "same");
- writeFile("b.txt", contents: "after");
- expectModifyEvent("b.txt");
- });
-
- test('does not notify if the modification time did not change', () {
- writeFile("a.txt", contents: "before");
- writeFile("b.txt", contents: "before");
- createWatcher();
- writeFile("a.txt", contents: "after", updateModified: false);
- writeFile("b.txt", contents: "after");
- expectModifyEvent("b.txt");
- });
-
- test('watches files in subdirectories', () {
- createWatcher();
- writeFile("a/b/c/d/file.txt");
- expectAddEvent("a/b/c/d/file.txt");
- });
-
- test('watches a directory created after the watcher', () {
- // Watch a subdirectory that doesn't exist yet.
- createWatcher(dir: "a");
-
- // This implicity creates it.
- writeFile("a/b/c/d/file.txt");
- expectAddEvent("a/b/c/d/file.txt");
- });
-
- test('when the watched directory is deleted, removes all files', () {
- writeFile("dir/a.txt");
- writeFile("dir/b.txt");
-
- createWatcher(dir: "dir");
-
- deleteDir("dir");
- expectRemoveEvents(["dir/a.txt", "dir/b.txt"]);
- });
-}
diff --git a/pkgs/watcher/test/no_subscription/linux_test.dart b/pkgs/watcher/test/no_subscription/linux_test.dart
new file mode 100644
index 0000000..7978830
--- /dev/null
+++ b/pkgs/watcher/test/no_subscription/linux_test.dart
@@ -0,0 +1,20 @@
+// Copyright (c) 2013, the Dart project authors. Please see the AUTHORS file
+// for details. All rights reserved. Use of this source code is governed by a
+// BSD-style license that can be found in the LICENSE file.
+
+import 'package:scheduled_test/scheduled_test.dart';
+import 'package:watcher/src/directory_watcher/linux.dart';
+import 'package:watcher/watcher.dart';
+
+import 'shared.dart';
+import '../utils.dart';
+
+main() {
+ initConfig();
+
+ watcherFactory = (dir) => new LinuxDirectoryWatcher(dir);
+
+ setUp(createSandbox);
+
+ sharedTests();
+}
\ No newline at end of file
diff --git a/pkgs/watcher/test/no_subscription/polling_test.dart b/pkgs/watcher/test/no_subscription/polling_test.dart
new file mode 100644
index 0000000..fa4f0cb
--- /dev/null
+++ b/pkgs/watcher/test/no_subscription/polling_test.dart
@@ -0,0 +1,19 @@
+// Copyright (c) 2013, the Dart project authors. Please see the AUTHORS file
+// for details. All rights reserved. Use of this source code is governed by a
+// BSD-style license that can be found in the LICENSE file.
+
+import 'package:scheduled_test/scheduled_test.dart';
+import 'package:watcher/watcher.dart';
+
+import 'shared.dart';
+import '../utils.dart';
+
+main() {
+ initConfig();
+
+ watcherFactory = (dir) => new PollingDirectoryWatcher(dir);
+
+ setUp(createSandbox);
+
+ sharedTests();
+}
\ No newline at end of file
diff --git a/pkgs/watcher/test/no_subscription_test.dart b/pkgs/watcher/test/no_subscription/shared.dart
similarity index 73%
rename from pkgs/watcher/test/no_subscription_test.dart
rename to pkgs/watcher/test/no_subscription/shared.dart
index 2e7b6d3..cd279e1 100644
--- a/pkgs/watcher/test/no_subscription_test.dart
+++ b/pkgs/watcher/test/no_subscription/shared.dart
@@ -7,14 +7,10 @@
import 'package:scheduled_test/scheduled_test.dart';
import 'package:watcher/watcher.dart';
-import 'utils.dart';
+import '../utils.dart';
-main() {
- initConfig();
-
- setUp(createSandbox);
-
- test('does not notify for changes when there were no subscribers', () {
+sharedTests() {
+ test('does not notify for changes when there are no subscribers', () {
// Note that this test doesn't rely as heavily on the test functions in
// utils.dart because it needs to be very explicit about when the event
// stream is and is not subscribed.
@@ -51,16 +47,10 @@
expect(event.path, endsWith("added.txt"));
completer.complete();
}));
- });
- // The watcher will have been cancelled and then resumed in the middle of
- // its pause between polling loops. That means the second scan to skip
- // what changed while we were unsubscribed won't happen until after that
- // delay is done. Wait long enough for that to happen.
- //
- // We're doing * 4 here because that seems to give the slower bots enough
- // time for this to complete.
- schedule(() => new Future.delayed(watcher.pollingDelay * 4));
+ // Wait until the watcher is ready to dispatch events again.
+ return watcher.ready;
+ });
// And add a third file.
writeFile("added.txt");
diff --git a/pkgs/watcher/test/ready/linux_test.dart b/pkgs/watcher/test/ready/linux_test.dart
new file mode 100644
index 0000000..7978830
--- /dev/null
+++ b/pkgs/watcher/test/ready/linux_test.dart
@@ -0,0 +1,20 @@
+// Copyright (c) 2013, the Dart project authors. Please see the AUTHORS file
+// for details. All rights reserved. Use of this source code is governed by a
+// BSD-style license that can be found in the LICENSE file.
+
+import 'package:scheduled_test/scheduled_test.dart';
+import 'package:watcher/src/directory_watcher/linux.dart';
+import 'package:watcher/watcher.dart';
+
+import 'shared.dart';
+import '../utils.dart';
+
+main() {
+ initConfig();
+
+ watcherFactory = (dir) => new LinuxDirectoryWatcher(dir);
+
+ setUp(createSandbox);
+
+ sharedTests();
+}
\ No newline at end of file
diff --git a/pkgs/watcher/test/ready/polling_test.dart b/pkgs/watcher/test/ready/polling_test.dart
new file mode 100644
index 0000000..fa4f0cb
--- /dev/null
+++ b/pkgs/watcher/test/ready/polling_test.dart
@@ -0,0 +1,19 @@
+// Copyright (c) 2013, the Dart project authors. Please see the AUTHORS file
+// for details. All rights reserved. Use of this source code is governed by a
+// BSD-style license that can be found in the LICENSE file.
+
+import 'package:scheduled_test/scheduled_test.dart';
+import 'package:watcher/watcher.dart';
+
+import 'shared.dart';
+import '../utils.dart';
+
+main() {
+ initConfig();
+
+ watcherFactory = (dir) => new PollingDirectoryWatcher(dir);
+
+ setUp(createSandbox);
+
+ sharedTests();
+}
\ No newline at end of file
diff --git a/pkgs/watcher/test/ready_test.dart b/pkgs/watcher/test/ready/shared.dart
similarity index 95%
rename from pkgs/watcher/test/ready_test.dart
rename to pkgs/watcher/test/ready/shared.dart
index 11b77e0..af1b58f 100644
--- a/pkgs/watcher/test/ready_test.dart
+++ b/pkgs/watcher/test/ready/shared.dart
@@ -2,17 +2,11 @@
// for details. All rights reserved. Use of this source code is governed by a
// BSD-style license that can be found in the LICENSE file.
-import 'dart:async';
-
import 'package:scheduled_test/scheduled_test.dart';
-import 'utils.dart';
+import '../utils.dart';
-main() {
- initConfig();
-
- setUp(createSandbox);
-
+sharedTests() {
test('ready does not complete until after subscription', () {
var watcher = createWatcher(waitForReady: false);
diff --git a/pkgs/watcher/test/utils.dart b/pkgs/watcher/test/utils.dart
index 28d57b0..d1b9f94 100644
--- a/pkgs/watcher/test/utils.dart
+++ b/pkgs/watcher/test/utils.dart
@@ -5,6 +5,7 @@
library watcher.test.utils;
import 'dart:async';
+import 'dart:collection';
import 'dart:io';
import 'package:path/path.dart' as p;
@@ -12,6 +13,7 @@
import 'package:unittest/compact_vm_config.dart';
import 'package:watcher/watcher.dart';
import 'package:watcher/src/stat.dart';
+import 'package:watcher/src/utils.dart';
/// The path to the temporary sandbox created for each test. All file
/// operations are implicitly relative to this directory.
@@ -36,6 +38,14 @@
/// increment the mod time for that file instantly.
Map<String, int> _mockFileModificationTimes;
+typedef DirectoryWatcher WatcherFactory(String directory);
+
+/// Sets the function used to create the directory watcher.
+set watcherFactory(WatcherFactory factory) {
+ _watcherFactory = factory;
+}
+WatcherFactory _watcherFactory;
+
void initConfig() {
useCompactVMConfiguration();
filterStacks = true;
@@ -54,10 +64,10 @@
path = p.normalize(p.relative(path, from: _sandboxDir));
// Make sure we got a path in the sandbox.
- assert(p.isRelative(path) && !path.startsWith(".."));
+ assert(p.isRelative(path) && !path.startsWith(".."));
- return new DateTime.fromMillisecondsSinceEpoch(
- _mockFileModificationTimes[path]);
+ var mtime = _mockFileModificationTimes[path];
+ return new DateTime.fromMillisecondsSinceEpoch(mtime == null ? 0 : mtime);
});
// Delete the sandbox when done.
@@ -86,68 +96,113 @@
dir = p.join(_sandboxDir, dir);
}
- // Use a short delay to make the tests run quickly.
- _watcher = new DirectoryWatcher(dir,
- pollingDelay: new Duration(milliseconds: 100));
+ var watcher = _watcherFactory(dir);
// Wait until the scan is finished so that we don't miss changes to files
// that could occur before the scan completes.
if (waitForReady != false) {
- schedule(() => _watcher.ready, "wait for watcher to be ready");
+ schedule(() => watcher.ready, "wait for watcher to be ready");
}
- currentSchedule.onComplete.schedule(() {
- _nextEvent = 0;
- _watcher = null;
- }, "reset watcher");
-
- return _watcher;
+ return watcher;
}
-/// Expects that the next set of events will all be changes of [type] on
-/// [paths].
+/// The stream of events from the watcher started with [startWatcher].
+Stream _watcherEvents;
+
+/// Creates a new [DirectoryWatcher] that watches a temporary directory and
+/// starts monitoring it for events.
///
-/// Validates that events are delivered for all paths in [paths], but allows
-/// them in any order.
-void expectEvents(ChangeType type, Iterable<String> paths) {
- var pathSet = paths
- .map((path) => p.join(_sandboxDir, path))
- .map(p.normalize)
- .toSet();
+/// If [dir] is provided, watches a subdirectory in the sandbox with that name.
+void startWatcher({String dir}) {
+ // We want to wait until we're ready *after* we subscribe to the watcher's
+ // events.
+ _watcher = createWatcher(dir: dir, waitForReady: false);
- // Create an expectation for as many paths as we have.
- var futures = [];
+ // Schedule [_watcher.events.listen] so that the watcher doesn't start
+ // watching [dir] before it exists. Expose [_watcherEvents] immediately so
+ // that it can be accessed synchronously after this.
+ _watcherEvents = futureStream(schedule(() {
+ var allEvents = new Queue();
+ var subscription = _watcher.events.listen(allEvents.add,
+ onError: currentSchedule.signalError);
- for (var i = 0; i < paths.length; i++) {
- // Immediately create the futures. This ensures we don't register too
- // late and drop the event before we receive it.
- var future = _watcher.events.elementAt(_nextEvent++).then((event) {
- expect(event.type, equals(type));
- expect(pathSet, contains(event.path));
+ currentSchedule.onComplete.schedule(() {
+ var numEvents = _nextEvent;
+ subscription.cancel();
+ _nextEvent = 0;
+ _watcher = null;
- pathSet.remove(event.path);
- });
+ // If there are already errors, don't add this to the output and make
+ // people think it might be the root cause.
+ if (currentSchedule.errors.isEmpty) {
+ expect(allEvents, hasLength(numEvents));
+ }
+ }, "reset watcher");
- // Make sure the schedule is watching it in case it fails.
- currentSchedule.wrapFuture(future);
+ return _watcher.events;
+ }, "create watcher")).asBroadcastStream();
- futures.add(future);
+ schedule(() => _watcher.ready, "wait for watcher to be ready");
+}
+
+/// A future set by [inAnyOrder] that will complete to the set of events that
+/// occur in the [inAnyOrder] block.
+Future<Set<WatchEvent>> _unorderedEventFuture;
+
+/// Runs [block] and allows multiple [expectEvent] calls in that block to match
+/// events in any order.
+void inAnyOrder(block()) {
+ var oldFuture = _unorderedEventFuture;
+ try {
+ var firstEvent = _nextEvent;
+ var completer = new Completer();
+ _unorderedEventFuture = completer.future;
+ block();
+
+ _watcherEvents.skip(firstEvent).take(_nextEvent - firstEvent).toSet()
+ .then(completer.complete, onError: completer.completeError);
+ currentSchedule.wrapFuture(_unorderedEventFuture,
+ "waiting for ${_nextEvent - firstEvent} events");
+ } finally {
+ _unorderedEventFuture = oldFuture;
}
-
- // Schedule it so that later file modifications don't occur until after this
- // event is received.
- schedule(() => Future.wait(futures),
- "wait for $type events on ${paths.join(', ')}");
}
-void expectAddEvent(String path) => expectEvents(ChangeType.ADD, [path]);
-void expectModifyEvent(String path) => expectEvents(ChangeType.MODIFY, [path]);
-void expectRemoveEvent(String path) => expectEvents(ChangeType.REMOVE, [path]);
+/// Expects that the next set of event will be a change of [type] on [path].
+///
+/// Multiple calls to [expectEvent] require that the events are received in that
+/// order unless they're called in an [inAnyOrder] block.
+void expectEvent(ChangeType type, String path) {
+ var matcher = predicate((e) {
+ return e is WatchEvent && e.type == type &&
+ e.path == p.join(_sandboxDir, path);
+ }, "is $type $path");
-void expectRemoveEvents(Iterable<String> paths) {
- expectEvents(ChangeType.REMOVE, paths);
+ if (_unorderedEventFuture != null) {
+ // Assign this to a local variable since it will be un-assigned by the time
+ // the scheduled callback runs.
+ var future = _unorderedEventFuture;
+
+ expect(
+ schedule(() => future, "should fire $type event on $path"),
+ completion(contains(matcher)));
+ } else {
+ var future = currentSchedule.wrapFuture(
+ _watcherEvents.elementAt(_nextEvent),
+ "waiting for $type event on $path");
+
+ expect(
+ schedule(() => future, "should fire $type event on $path"),
+ completion(matcher));
+ }
+ _nextEvent++;
}
+void expectAddEvent(String path) => expectEvent(ChangeType.ADD, path);
+void expectModifyEvent(String path) => expectEvent(ChangeType.MODIFY, path);
+void expectRemoveEvent(String path) => expectEvent(ChangeType.REMOVE, path);
+
/// Schedules writing a file in the sandbox at [path] with [contents].
///
/// If [contents] is omitted, creates an empty file. If [updatedModified] is
@@ -201,6 +256,21 @@
}, "rename file $from to $to");
}
+/// Schedules creating a directory in the sandbox at [path].
+void createDir(String path) {
+ schedule(() {
+ new Directory(p.join(_sandboxDir, path)).createSync();
+ }, "create directory $path");
+}
+
+/// Schedules renaming a directory in the sandbox from [from] to [to].
+void renameDir(String from, String to) {
+ schedule(() {
+ new Directory(p.join(_sandboxDir, from))
+ .renameSync(p.join(_sandboxDir, to));
+ }, "rename directory $from to $to");
+}
+
/// Schedules deleting a directory in the sandbox at [path].
void deleteDir(String path) {
schedule(() {
@@ -208,22 +278,17 @@
}, "delete directory $path");
}
-/// A [Matcher] for [WatchEvent]s.
-class _ChangeMatcher extends Matcher {
- /// The expected change.
- final ChangeType type;
-
- /// The expected path.
- final String path;
-
- _ChangeMatcher(this.type, this.path);
-
- Description describe(Description description) {
- description.add("$type $path");
+/// Runs [callback] with every permutation of non-negative [i], [j], and [k]
+/// less than [limit].
+///
+/// [limit] defaults to 3.
+void withPermutations(callback(int i, int j, int k), {int limit}) {
+ if (limit == null) limit = 3;
+ for (var i = 0; i < limit; i++) {
+ for (var j = 0; j < limit; j++) {
+ for (var k = 0; k < limit; k++) {
+ callback(i, j, k);
+ }
+ }
}
-
- bool matches(item, Map matchState) =>
- item is WatchEvent &&
- item.type == type &&
- p.normalize(item.path) == p.normalize(path);
}
