File watching package. BUG= R=nweiz@google.com Review URL: https://codereview.chromium.org//18612013 git-svn-id: https://dart.googlecode.com/svn/branches/bleeding_edge/dart/pkg/watcher@24971 260f80e4-7a28-3924-810f-c04153c831b5
diff --git a/pkgs/watcher/README.md b/pkgs/watcher/README.md new file mode 100644 index 0000000..75b0470 --- /dev/null +++ b/pkgs/watcher/README.md
@@ -0,0 +1,2 @@ +A file watcher. It monitors (currently by polling) for changes to contents of +directories and notifies you when files have been added, removed, or modified. \ No newline at end of file
diff --git a/pkgs/watcher/example/watch.dart b/pkgs/watcher/example/watch.dart new file mode 100644 index 0000000..247f66a --- /dev/null +++ b/pkgs/watcher/example/watch.dart
@@ -0,0 +1,24 @@ +// 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. + +/// Watches the given directory and prints each modification to it. +library watch; + +import 'dart:io'; + +import 'package:pathos/path.dart' as pathos; +import 'package:watcher/watcher.dart'; + +main() { + var args = new Options().arguments; + if (args.length != 1) { + print("Usage: watch <directory path>"); + return; + } + + var watcher = new DirectoryWatcher(pathos.absolute(args[0])); + watcher.events.listen((event) { + print(event); + }); +} \ No newline at end of file
diff --git a/pkgs/watcher/lib/src/directory_watcher.dart b/pkgs/watcher/lib/src/directory_watcher.dart new file mode 100644 index 0000000..0f297ba --- /dev/null +++ b/pkgs/watcher/lib/src/directory_watcher.dart
@@ -0,0 +1,226 @@ +// 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; + +import 'dart:async'; +import 'dart:io'; + +import 'package:crypto/crypto.dart'; + +import 'stat.dart'; +import 'watch_event.dart'; + +/// Watches the contents of a directory and emits [WatchEvent]s when something +/// in the directory has changed. +class DirectoryWatcher { + /// The directory whose contents are being monitored. + final String directory; + + /// The broadcast [Stream] of events that have occurred to files in + /// [directory]. + /// + /// 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; + + _WatchState _state = _WatchState.notWatching; + + /// A [Future] that completes when the watcher is initialized and watching + /// for file changes. + /// + /// If the watcher is not currently monitoring the directory (because there + /// are no subscribers to [events]), this returns a future that isn't + /// complete yet. It will complete when a subscriber starts listening and + /// the watcher finishes any initialization work it needs to do. + /// + /// If the watcher is already monitoring, this returns an already complete + /// future. + Future get ready => _ready.future; + Completer _ready = new Completer(); + + /// The previous status of the files in the directory. + /// + /// Used to tell which files have been modified. + final _statuses = new Map<String, _FileStatus>(); + + /// Creates a new [DirectoryWatcher] monitoring [directory]. + DirectoryWatcher(this.directory) { + _events = new StreamController<WatchEvent>.broadcast(onListen: () { + _state = _state.listen(this); + }, onCancel: () { + _state = _state.cancel(this); + }); + } + + /// Starts the asynchronous polling process. + /// + /// Scans the contents of the directory and compares the results to the + /// previous scan. Loops to continue monitoring as long as there are + /// subscribers to the [events] stream. + Future _watch() { + var files = new Set<String>(); + + var stream = new Directory(directory).list(recursive: true); + + return stream.map((entity) { + if (entity is! File) return new Future.value(); + files.add(entity.path); + // TODO(rnystrom): These all run as fast as possible and read the + // contents of the files. That means there's a pretty big IO hit all at + // once. Maybe these should be queued up and rate limited? + return _refreshFile(entity.path); + }).toList().then((futures) { + // Once the listing is done, make sure to wait until each file is also + // done. + return Future.wait(futures); + }).then((_) { + var removedFiles = _statuses.keys.toSet().difference(files); + for (var removed in removedFiles) { + if (_state.shouldNotify) { + _events.add(new WatchEvent(ChangeType.REMOVE, removed)); + } + _statuses.remove(removed); + } + + var previousState = _state; + _state = _state.finish(this); + + // If we were already sending notifications, add a bit of delay before + // restarting just so that we don't whale on the file system. + // TODO(rnystrom): Tune this and/or make it tunable? + if (_state.shouldNotify) { + return new Future.delayed(new Duration(seconds: 1)); + } + }).then((_) { + // Make sure we haven't transitioned to a non-watching state during the + // delay. + if (_state.shouldWatch) _watch(); + }); + } + + /// Compares the current state of the file at [path] to the state it was in + /// the last time it was scanned. + Future _refreshFile(String path) { + return getModificationTime(path).then((modified) { + var lastStatus = _statuses[path]; + + // If it's modification time hasn't changed, assume the file is unchanged. + if (lastStatus != null && lastStatus.modified == modified) return; + + return _hashFile(path).then((hash) { + var status = new _FileStatus(modified, hash); + _statuses[path] = status; + + // Only notify if the file contents changed. + if (_state.shouldNotify && + (lastStatus == null || !_sameHash(lastStatus.hash, hash))) { + var change = lastStatus == null ? ChangeType.ADD : ChangeType.MODIFY; + _events.add(new WatchEvent(change, path)); + } + }); + }); + } + + /// 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; + } +} + +/// An "event" that is sent to the [_WatchState] FSM to trigger state +/// transitions. +typedef _WatchState _WatchStateEvent(DirectoryWatcher watcher); + +/// The different states that the watcher can be in and the transitions between +/// them. +/// +/// This class defines a finite state machine for keeping track of what the +/// asynchronous file polling is doing. Each instance of this is a state in the +/// machine and its [listen], [cancel], and [finish] fields define the state +/// transitions when those events occur. +class _WatchState { + /// The watcher has no subscribers. + static final notWatching = new _WatchState( + listen: (watcher) { + watcher._watch(); + return _WatchState.scanning; + }); + + /// The watcher has subscribers and is scanning for pre-existing files. + static final scanning = new _WatchState( + cancel: (watcher) { + // No longer watching, so create a new incomplete ready future. + watcher._ready = new Completer(); + return _WatchState.cancelling; + }, finish: (watcher) { + watcher._ready.complete(); + return _WatchState.watching; + }, shouldWatch: true); + + /// The watcher was unsubscribed while polling and we're waiting for the poll + /// to finish. + static final cancelling = new _WatchState( + listen: (_) => _WatchState.scanning, + finish: (_) => _WatchState.notWatching); + + /// The watcher has subscribers, we have scanned for pre-existing files and + /// now we're polling for changes. + static final watching = new _WatchState( + cancel: (watcher) { + // No longer watching, so create a new incomplete ready future. + watcher._ready = new Completer(); + return _WatchState.cancelling; + }, finish: (_) => _WatchState.watching, + shouldWatch: true, shouldNotify: true); + + /// Called when the first subscriber to the watcher has been added. + final _WatchStateEvent listen; + + /// Called when all subscriptions on the watcher have been cancelled. + final _WatchStateEvent cancel; + + /// Called when a poll loop has finished. + final _WatchStateEvent finish; + + /// If the directory watcher should be watching the file system while in + /// this state. + final bool shouldWatch; + + /// If a change event should be sent for a file modification while in this + /// state. + final bool shouldNotify; + + _WatchState({this.listen, this.cancel, this.finish, + this.shouldWatch: false, this.shouldNotify: false}); +} + +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/stat.dart b/pkgs/watcher/lib/src/stat.dart new file mode 100644 index 0000000..d36eff3 --- /dev/null +++ b/pkgs/watcher/lib/src/stat.dart
@@ -0,0 +1,33 @@ +// 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.stat; + +import 'dart:async'; +import 'dart:io'; + +/// A function that takes a file path and returns the last modified time for +/// the file at that path. +typedef DateTime MockTimeCallback(String path); + +MockTimeCallback _mockTimeCallback; + +/// Overrides the default behavior for accessing a file's modification time +/// with [callback]. +/// +/// The OS file modification time has pretty rough granularity (like a few +/// seconds) which can make for slow tests that rely on modtime. This lets you +/// replace it with something you control. +void mockGetModificationTime(MockTimeCallback callback) { + _mockTimeCallback = callback; +} + +/// Gets the modification time for the file at [path]. +Future<DateTime> getModificationTime(String path) { + if (_mockTimeCallback != null) { + return new Future.value(_mockTimeCallback(path)); + } + + return FileStat.stat(path).then((stat) => stat.modified); +}
diff --git a/pkgs/watcher/lib/src/watch_event.dart b/pkgs/watcher/lib/src/watch_event.dart new file mode 100644 index 0000000..d998a25 --- /dev/null +++ b/pkgs/watcher/lib/src/watch_event.dart
@@ -0,0 +1,35 @@ +// 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.watch_event; + +/// An event describing a single change to the file system. +class WatchEvent { + /// The manner in which the file at [path] has changed. + final ChangeType type; + + /// The path of the file that changed. + final String path; + + WatchEvent(this.type, this.path); + + String toString() => "$type $path"; +} + +/// Enum for what kind of change has happened to a file. +class ChangeType { + /// A new file has been added. + static const ADD = const ChangeType("add"); + + /// A file has been removed. + static const REMOVE = const ChangeType("remove"); + + /// The contents of a file have changed. + static const MODIFY = const ChangeType("modify"); + + final String _name; + const ChangeType(this._name); + + String toString() => _name; +} \ No newline at end of file
diff --git a/pkgs/watcher/lib/watcher.dart b/pkgs/watcher/lib/watcher.dart new file mode 100644 index 0000000..c4824b8 --- /dev/null +++ b/pkgs/watcher/lib/watcher.dart
@@ -0,0 +1,8 @@ +// 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; + +export 'src/watch_event.dart'; +export 'src/directory_watcher.dart';
diff --git a/pkgs/watcher/pubspec.yaml b/pkgs/watcher/pubspec.yaml new file mode 100644 index 0000000..263832e --- /dev/null +++ b/pkgs/watcher/pubspec.yaml
@@ -0,0 +1,13 @@ +name: watcher +author: "Dart Team <misc@dartlang.org>" +homepage: http://www.dartlang.org +description: > + A file watcher. It monitors (currently by polling) for changes to contents + of directories and notifies you when files have been added, removed, or + modified. +dependencies: + crypto: any + path: any +dev_dependencies: + scheduled_test: any + unittest: any
diff --git a/pkgs/watcher/test/directory_watcher_test.dart b/pkgs/watcher/test/directory_watcher_test.dart new file mode 100644 index 0000000..635f7ee --- /dev/null +++ b/pkgs/watcher/test/directory_watcher_test.dart
@@ -0,0 +1,239 @@ +// 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 'dart:async'; +import 'dart:io'; + +import 'package:scheduled_test/scheduled_test.dart'; +import 'package:watcher/watcher.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('does not notify for changes when there were 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. + var watcher = createWatcher(); + + // Subscribe to the events. + var completer = new Completer(); + var subscription = watcher.events.listen((event) { + expect(event.type, equals(ChangeType.ADD)); + expect(event.path, endsWith("file.txt")); + completer.complete(); + }); + + writeFile("file.txt"); + + // Then wait until we get an event for it. + schedule(() => completer.future); + + // Unsubscribe. + schedule(() { + subscription.cancel(); + }); + + // Now write a file while we aren't listening. + writeFile("unwatched.txt"); + + // Then start listening again. + schedule(() { + completer = new Completer(); + subscription = watcher.events.listen((event) { + // We should get an event for the third file, not the one added while + // we weren't subscribed. + expect(event.type, equals(ChangeType.ADD)); + 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. + schedule(() => new Future.delayed(new Duration(seconds: 1))); + + // And add a third file. + writeFile("added.txt"); + + // Wait until we get an event for the third file. + schedule(() => completer.future); + + schedule(() { + subscription.cancel(); + }); + }); + + + test('ready does not complete until after subscription', () { + var watcher = createWatcher(waitForReady: false); + + var ready = false; + watcher.ready.then((_) { + ready = true; + }); + + // Should not be ready yet. + schedule(() { + expect(ready, isFalse); + }); + + // Subscribe to the events. + schedule(() { + var subscription = watcher.events.listen((event) {}); + + currentSchedule.onComplete.schedule(() { + subscription.cancel(); + }); + }); + + // Should eventually be ready. + schedule(() => watcher.ready); + + schedule(() { + expect(ready, isTrue); + }); + }); + + test('ready completes immediately when already ready', () { + var watcher = createWatcher(waitForReady: false); + + // Subscribe to the events. + schedule(() { + var subscription = watcher.events.listen((event) {}); + + currentSchedule.onComplete.schedule(() { + subscription.cancel(); + }); + }); + + // Should eventually be ready. + schedule(() => watcher.ready); + + // Now ready should be a future that immediately completes. + var ready = false; + schedule(() { + watcher.ready.then((_) { + ready = true; + }); + }); + + schedule(() { + expect(ready, isTrue); + }); + }); + + test('ready returns a future that does not complete after unsubscribing', () { + var watcher = createWatcher(waitForReady: false); + + // Subscribe to the events. + var subscription; + schedule(() { + subscription = watcher.events.listen((event) {}); + }); + + var ready = false; + + // Wait until ready. + schedule(() => watcher.ready); + + // Now unsubscribe. + schedule(() { + subscription.cancel(); + + // Track when it's ready again. + ready = false; + watcher.ready.then((_) { + ready = true; + }); + }); + + // Should be back to not ready. + schedule(() { + expect(ready, isFalse); + }); + }); +}
diff --git a/pkgs/watcher/test/utils.dart b/pkgs/watcher/test/utils.dart new file mode 100644 index 0000000..387b4ad --- /dev/null +++ b/pkgs/watcher/test/utils.dart
@@ -0,0 +1,186 @@ +// 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. + +library watcher.test.utils; + +import 'dart:async'; +import 'dart:io'; + +import 'package:path/path.dart'; +import 'package:scheduled_test/scheduled_test.dart'; +import 'package:unittest/compact_vm_config.dart'; +import 'package:watcher/watcher.dart'; +import 'package:watcher/src/stat.dart'; + +/// The path to the temporary sandbox created for each test. All file +/// operations are implicitly relative to this directory. +String _sandboxDir; + +/// The [DirectoryWatcher] being used for the current scheduled test. +DirectoryWatcher _watcher; + +/// The index in [_watcher]'s event stream for the next event. When event +/// expectations are set using [expectEvent] (et. al.), they use this to +/// expect a series of events in order. +var _nextEvent = 0; + +/// The mock modification times (in milliseconds since epoch) for each file. +/// +/// The actual file system has pretty coarse granularity for file modification +/// times. This means using the real file system requires us to put delays in +/// the tests to ensure we wait long enough between operations for the mod time +/// to be different. +/// +/// Instead, we'll just mock that out. Each time a file is written, we manually +/// increment the mod time for that file instantly. +Map<String, int> _mockFileModificationTimes; + +void initConfig() { + useCompactVMConfiguration(); +} + +/// Creates the sandbox directory the other functions in this library use and +/// ensures it's deleted when the test ends. +/// +/// This should usually be called by [setUp]. +void createSandbox() { + var dir = new Directory("").createTempSync(); + _sandboxDir = dir.path; + + _mockFileModificationTimes = new Map<String, int>(); + mockGetModificationTime((path) { + path = relative(path, from: _sandboxDir); + + // Make sure we got a path in the sandbox. + assert(isRelative(path) && !path.startsWith("..")); + + return new DateTime.fromMillisecondsSinceEpoch( + _mockFileModificationTimes[path]); + }); + + // Delete the sandbox when done. + currentSchedule.onComplete.schedule(() { + if (_sandboxDir != null) { + new Directory(_sandboxDir).deleteSync(recursive: true); + _sandboxDir = null; + } + + _mockFileModificationTimes = null; + mockGetModificationTime(null); + }, "delete sandbox"); +} + +/// Creates a new [DirectoryWatcher] that watches a temporary directory. +/// +/// Normally, this will pause the schedule until the watcher is done scanning +/// and is polling for changes. If you pass `false` for [waitForReady], it will +/// not schedule this delay. +DirectoryWatcher createWatcher({bool waitForReady}) { + _watcher = new DirectoryWatcher(_sandboxDir); + + // 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); + } + + currentSchedule.onComplete.schedule(() { + _nextEvent = 0; + _watcher = null; + }, "reset watcher"); + + return _watcher; +} + +void expectEvent(ChangeType type, String path) { + // Immediately create the future. 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, new _ChangeMatcher(type, path)); + }); + + // Make sure the schedule is watching it in case it fails. + currentSchedule.wrapFuture(future); + + // Schedule it so that later file modifications don't occur until after this + // event is received. + schedule(() => future); +} + +void expectAddEvent(String path) { + expectEvent(ChangeType.ADD, join(_sandboxDir, path)); +} + +void expectModifyEvent(String path) { + expectEvent(ChangeType.MODIFY, join(_sandboxDir, path)); +} + +void expectRemoveEvent(String path) { + expectEvent(ChangeType.REMOVE, join(_sandboxDir, path)); +} + +/// Schedules writing a file in the sandbox at [path] with [contents]. +/// +/// If [contents] is omitted, creates an empty file. If [updatedModified] is +/// `false`, the mock file modification time is not changed. +void writeFile(String path, {String contents, bool updateModified}) { + if (contents == null) contents = ""; + if (updateModified == null) updateModified = true; + + schedule(() { + var fullPath = join(_sandboxDir, path); + + // Create any needed subdirectories. + var dir = new Directory(dirname(fullPath)); + if (!dir.existsSync()) { + dir.createSync(recursive: true); + } + + new File(fullPath).writeAsStringSync(contents); + + // Manually update the mock modification time for the file. + if (updateModified) { + var milliseconds = _mockFileModificationTimes.putIfAbsent(path, () => 0); + _mockFileModificationTimes[path]++; + } + }); +} + +/// Schedules deleting a file in the sandbox at [path]. +void deleteFile(String path) { + schedule(() { + new File(join(_sandboxDir, path)).deleteSync(); + }); +} + +/// Schedules renaming a file in the sandbox from [from] to [to]. +/// +/// If [contents] is omitted, creates an empty file. +void renameFile(String from, String to) { + schedule(() { + new File(join(_sandboxDir, from)).renameSync(join(_sandboxDir, to)); + + // Manually update the mock modification time for the file. + var milliseconds = _mockFileModificationTimes.putIfAbsent(to, () => 0); + _mockFileModificationTimes[to]++; + }); +} + +/// A [Matcher] for [WatchEvent]s. +class _ChangeMatcher extends BaseMatcher { + /// 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"); + } + + bool matches(item, Map matchState) => + item is WatchEvent && item.type == type && item.path == path; +}