Wrap Directory.watch on linux for the watcher package.

pkg/watcher/lib/src/directory_watcher/polling.dart

 // 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;
+library watcher.directory_watcher.polling;
 
 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 '../async_queue.dart';
+import '../directory_watcher.dart';
+import '../stat.dart';
+import '../utils.dart';
+import '../watch_event.dart';
+import 'resubscribable.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.
+/// 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;
 
-  /// 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;
+  final _events = new StreamController<WatchEvent>.broadcast();
 
-  _WatchState _state = _WatchState.UNSUBSCRIBED;
+  bool get isReady => _ready.isCompleted;
 
-  /// 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();
+  final _ready = new Completer();
 
   /// The amount of time the watcher pauses between successive polls of the
   /// directory contents.
-  final Duration pollingDelay;
+  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>();
 
-  /// 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);
-
+  _PollingDirectoryWatcher(this.directory, this._pollingDelay) {
     _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;
+  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();
-
-    _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);
+      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(_state != _WatchState.UNSUBSCRIBED);
+      assert(!_events.isClosed);
 
       if (entity is! File) return;
       _filesToProcess.add(entity.path);
-    }, onError: (error, StackTrace stackTrace) {
+    }, 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) {
-    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;
+      if (_events.isClosed) return;
 
       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;
       }
 
       return _hashFile(file).then((hash) {
-        if (_checkForCancel()) return;
+        if (_events.isClosed) 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;
+        // 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 (_state == _WatchState.WATCHING) {
-        _events.add(new WatchEvent(ChangeType.REMOVE, removed));
-      }
+      if (isReady) _events.add(new WatchEvent(ChangeType.REMOVE, removed));
       _statuses.remove(removed);
     }
 
-    if (_state == _WatchState.SCANNING) {
-      _state = _WatchState.WATCHING;
-      _ready.complete();
-    }
+    if (!isReady) _ready.complete();
 
     // Wait and then poll again.
-    return new Future.delayed(pollingDelay).then((_) {
-      if (_checkForCancel()) return;
+    return new Future.delayed(_pollingDelay).then((_) {
+      if (_events.isClosed) 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.
-  ///
-  /// 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);
-}
-
 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);
-}
+}
+