Rewrite file-system-watcher to better handle the different system APIs.

runtime/bin/file_patch.dart

Index: runtime/bin/file_patch.dart
diff --git a/runtime/bin/file_patch.dart b/runtime/bin/file_patch.dart
index 9f1f6d06aa165d8289fc49861e936487021335ca..523c92cae16128cdca94f3980e6d37826fbca727 100644
--- a/runtime/bin/file_patch.dart
+++ b/runtime/bin/file_patch.dart
@@ -20,6 +20,7 @@ patch class _File {
   /* patch */ static int _openStdio(int fd) native "File_OpenStdio";
 }
 
+
 patch class _RandomAccessFile {
   /* patch */ static int _close(int id) native "File_Close";
   /* patch */ static _readByte(int id) native "File_ReadByte";
@@ -37,52 +38,121 @@ patch class _RandomAccessFile {
   /* patch */ static _flush(int id) native "File_Flush";
 }
 
-patch class _FileSystemWatcher {
-  /* patch */ factory _FileSystemWatcher(
-      String path, int events, bool recursive)
-    => new _FileSystemWatcherImpl(path, events, recursive);
 
-  /* patch */ static bool get isSupported => _FileSystemWatcherImpl.isSupported;
+class _WatcherPath {
+  final int pathId;
+  final String path;
+  final int events;
+  int count = 0;
+  _WatcherPath(this.pathId, this.path, this.events);
 }
 
-class _FileSystemWatcherImpl
-    extends NativeFieldWrapperClass1
-    implements _FileSystemWatcher {
+
+patch class _FileSystemWatcher {
+  static int _id;
+  static final Map<int, _FileSystemWatcherPath> _idMap = {};
+
   final String _path;
   final int _events;
   final bool _recursive;
 
-  StreamController _controller;
-  StreamSubscription _subscription;
+  _WatcherPath _watcherPath;
+
+  StreamController _broadcastController;
+
+  /* patch */ static Stream<FileSystemEvent> watch(
+      String path, int events, bool recursive) {
+    if (Platform.isLinux) {
+      return new _InotifyFileSystemWatcher(path, events, recursive).stream;
+    }
+    if (Platform.isWindows) {
+      return new _Win32FileSystemWatcher(path, events, recursive).stream;
+    }
+    if (Platform.isMacOS) {
+      return new _FSEventStreamFileSystemWatcher(
+          path, events, recursive).stream;
+    }
+    throw new FileSystemException(
+        "File system watching is not supported on this platform");
+  }
 
-  _FileSystemWatcherImpl(this._path, this._events, this._recursive) {
+  _FileSystemWatcher._(this._path, this._events, this._recursive) {
     if (!isSupported) {
       throw new FileSystemException(
-          "File system watching is not supported on this system",
+          "File system watching is not supported on this platform",
           _path);
     }
-    _controller = new StreamController.broadcast(onListen: _listen,
-                                                 onCancel: _cancel);
+    _broadcastController = new StreamController.broadcast(onListen: _listen,
+                                                          onCancel: _cancel);
   }
 
+  Stream get stream => _broadcastController.stream;
+
   void _listen() {
-    int socketId;
+    if (_id == null) {
+      try {
+        _id = _initWatcher();
+        _newWatcher();
+      } catch (e) {
+        _broadcastController.addError(new FileSystemException(
+            "Failed to initialize file system entity watcher"));
+        _broadcastController.close();
+        return;
+      }
+    }
+    var pathId;
     try {
-      socketId = _watchPath(_path, _events, identical(true, _recursive));
+      pathId = _watchPath(_id, _path, _events, _recursive);
     } catch (e) {
-      _controller.addError(new FileSystemException(
+      _broadcastController.addError(new FileSystemException(
           "Failed to watch path", _path, e));
-      _controller.close();
+      _broadcastController.close();
       return;
     }
+    if (!_idMap.containsKey(pathId)) {
+      _idMap[pathId] = new _WatcherPath(pathId, _path, _events);
+    }
+    _watcherPath = _idMap[pathId];
+    _watcherPath.count++;
+    _pathWatched().pipe(_broadcastController);
+  }
+
+  void _cancel() {
+    if (_watcherPath != null) {
+      assert(_watcherPath.count > 0);
+      _watcherPath.count--;
+      if (_watcherPath.count == 0) {
+        _pathWatchedEnd();
+        _unwatchPath(_id, _watcherPath.pathId);
+        _idMap.remove(_watcherPath.pathId);
+      }
+      _watcherPath = null;
+    }
+    if (_idMap.isEmpty && _id != null) {
+      _closeWatcher(_id);
+      _doneWatcher();
+      _id = null;
+    }
+  }
+

[Søren Gjesse, 2013/12/03 14:42:19]

Please add a comment here on the lifecycle and how these four methods are
implemented in the subclasses.

[Anders Johnsen, 2013/12/04 10:30:52]

Done.

+  void _newWatcher();
+  void _doneWatcher();
+  Stream _pathWatched();
+  void _donePathWatched();
+
+  static _WatcherPath _pathFromPathId(int pathId) {
+    return _idMap[pathId];
+  }
+
+  static Stream _listenOnSocket(int socketId, int id, int pathId) {
     var socket = new _RawSocket(new _NativeSocket.watch(socketId));
-    _subscription = socket.expand((event) {
-      bool stop = false;
+    return socket.expand((event) {
+      var stops = [];
       var events = [];
       var pair = {};
       if (event == RawSocketEvent.READ) {
         String getPath(event) {
-          var path = _path;
+          var path = _pathFromPathId(event[4]).path;
           if (event[2] != null && event[2].isNotEmpty) {
             path += Platform.pathSeparator;
             path += event[2];
@@ -96,87 +166,202 @@ class _FileSystemWatcherImpl
           }
           return (event[0] & FileSystemEvent._IS_DIR) != 0;
         }
-        void add(event) {
-          if ((event.type & _events) == 0) return;
-          events.add(event);
+        void add(id, event) {
+          if ((event.type & _pathFromPathId(id).events) == 0) return;
+          events.add([id, event]);
         }
         void rewriteMove(event, isDir) {
           if (event[3]) {
-            add(new FileSystemCreateEvent._(getPath(event), isDir));
+            add(event[4], new FileSystemCreateEvent._(getPath(event), isDir));
           } else {
-            add(new FileSystemDeleteEvent._(getPath(event), isDir));
+            add(event[4], new FileSystemDeleteEvent._(getPath(event), isDir));
           }
         }
         while (socket.available() > 0) {
-          for (var event in _readEvents()) {
+          for (var event in _readEvents(id, pathId)) {
             if (event == null) continue;
+            int pathId = event[4];
             bool isDir = getIsDir(event);
             var path = getPath(event);
             if ((event[0] & FileSystemEvent.CREATE) != 0) {
-              add(new FileSystemCreateEvent._(path, isDir));
+              add(event[4], new FileSystemCreateEvent._(path, isDir));
             }
             if ((event[0] & FileSystemEvent.MODIFY) != 0) {
-              add(new FileSystemModifyEvent._(path, isDir, true));
+              add(event[4], new FileSystemModifyEvent._(path, isDir, true));
             }
             if ((event[0] & FileSystemEvent._MODIFY_ATTRIBUTES) != 0) {
-              add(new FileSystemModifyEvent._(path, isDir, false));
+              add(event[4], new FileSystemModifyEvent._(path, isDir, false));
             }
             if ((event[0] & FileSystemEvent.MOVE) != 0) {
               int link = event[1];
               if (link > 0) {
-                if (pair.containsKey(link)) {
-                  events.add(new FileSystemMoveEvent._(
-                      getPath(pair[link]), isDir, path));
-                  pair.remove(link);
+                pair.putIfAbsent(pathId, () => {});
+                if (pair[pathId].containsKey(link)) {
+                  add(event[4],
+                      new FileSystemMoveEvent._(
+                          getPath(pair[pathId][link]), isDir, path));
+                  pair[pathId].remove(link);
                 } else {
-                  pair[link] = event;
+                  pair[pathId][link] = event;
                 }
               } else {
                 rewriteMove(event, isDir);
               }
             }
             if ((event[0] & FileSystemEvent.DELETE) != 0) {
-              add(new FileSystemDeleteEvent._(path, isDir));
+              add(event[4], new FileSystemDeleteEvent._(path, isDir));
             }
             if ((event[0] & FileSystemEvent._DELETE_SELF) != 0) {
-              add(new FileSystemDeleteEvent._(path, isDir));
-              stop = true;
+              add(event[4], new FileSystemDeleteEvent._(path, isDir));
+              // Signal done event.
+              stops.add([event[4], null]);
             }
           }
         }
-        for (var event in pair.values) {
-          rewriteMove(event, getIsDir(event));
+        for (var map in pair.values) {
+          for (var event in map.values) {
+            rewriteMove(event, getIsDir(event));
+          }
         }
       } else if (event == RawSocketEvent.CLOSED) {
       } else if (event == RawSocketEvent.READ_CLOSED) {
       } else {
         assert(false);
       }
-      if (stop) socket.close();
+      events.addAll(stops);
       return events;
-    })
-    .listen(_controller.add, onDone: _cancel);
+    });
   }
 
-  void _cancel() {
-    if (_subscription != null) {
-      _unwatchPath();
-      _subscription.cancel();
-      _subscription = null;
+  /* patch */ static bool get isSupported
+      native "FileSystemWatcher_IsSupported";
+
+  static int _initWatcher() native "FileSystemWatcher_InitWatcher";
+  static void _closeWatcher(int id) native "FileSystemWatcher_CloseWatcher";
+
+  static int _watchPath(int id, String path, int events, bool recursive)
+      native "FileSystemWatcher_WatchPath";
+  static void _unwatchPath(int id, int path_id)
+      native "FileSystemWatcher_UnwatchPath";
+  static List _readEvents(int id, int path_id)
+      native "FileSystemWatcher_ReadEvents";
+  static int _getSocketId(int id, int path_id)
+      native "FileSystemWatcher_GetSocketId";
+}
+
+
+class _InotifyFileSystemWatcher extends _FileSystemWatcher {
+  static final Map<int, StreamController> _idMap = {};
+  static StreamSubscription _subscription;
+
+  _InotifyFileSystemWatcher(path, events, recursive)
+      : super._(path, events, recursive);
+
+  void _newWatcher() {
+    int id = _FileSystemWatcher._id;
+    _subscription = _FileSystemWatcher._listenOnSocket(id, id, 0)
+      .listen((event) {
+        if (_idMap.containsKey(event[0])) {
+          if (event[1] != null) {
+            _idMap[event[0]].add(event[1]);
+          } else {
+            _idMap[event[0]].close();
+          }
+        }
+      });
+  }
+
+  void _doneWatcher() {

[Søren Gjesse, 2013/12/03 14:42:19]

Also close the inotify fd here?

[Anders Johnsen, 2013/12/04 10:30:52]

canceling a one-way stream will close the fd.

+    _subscription.cancel();
+  }
+
+  Stream _pathWatched() {
+    var pathId = _watcherPath.pathId;
+    if (!_idMap.containsKey(pathId)) {
+      _idMap[pathId] = new StreamController.broadcast();
     }
+    return _idMap[pathId].stream;
+  }
+
+  void _pathWatchedEnd() {
+    var pathId = _watcherPath.pathId;
+    if (!_idMap.containsKey(pathId)) return;
+    _idMap[pathId].close();
+    _idMap.remove(pathId);
+  }
+}
+
+
+class _Win32FileSystemWatcher extends _FileSystemWatcher {
+  StreamSubscription _subscription;
+  StreamController _controller;
+
+  _Win32FileSystemWatcher(path, events, recursive)
+      : super._(path, events, recursive);
+
+  void _newWatcher() {

[Søren Gjesse, 2013/12/03 14:42:19]

You can move the closing curly one line up if you like. The empty implementation
could be in the superclass.

[Anders Johnsen, 2013/12/04 10:30:52]

Done.

+  }
+
+  void _doneWatcher() {
+  }
+
+  Stream _pathWatched() {
+    var pathId = _watcherPath.pathId;
+    _controller = new StreamController();
+    _subscription = _FileSystemWatcher._listenOnSocket(pathId, 0, pathId)
+        .listen((event) {
+          assert(event[0] == pathId);
+          if (event[1] != null) {
+            _controller.add(event[1]);
+          } else {
+            _controller.close();
+          }
+        });
+    return _controller.stream;
+  }
+
+  void _pathWatchedEnd() {
+    _subscription.cancel();
     _controller.close();
   }
+}
+
 
-  Stream<FileSystemEvent> get stream => _controller.stream;
+class _FSEventStreamFileSystemWatcher extends _FileSystemWatcher {
+  StreamSubscription _subscription;
+  StreamController _controller;
 
-  static bool get isSupported native "FileSystemWatcher_IsSupported";
+  _FSEventStreamFileSystemWatcher(path, events, recursive)
+      : super._(path, events, recursive);
 
-  int _watchPath(String path, int events, bool recursive)
-      native "FileSystemWatcher_WatchPath";
-  void _unwatchPath() native "FileSystemWatcher_UnwatchPath";
-  List _readEvents() native "FileSystemWatcher_ReadEvents";
+  void _newWatcher() {
+  }
+
+  void _doneWatcher() {
+  }
+
+  Stream _pathWatched() {
+    var pathId = _watcherPath.pathId;
+    var socketId = _FileSystemWatcher._getSocketId(0, pathId);
+    _controller = new StreamController();
+    _subscription = _FileSystemWatcher._listenOnSocket(socketId, 0, pathId)
+        .listen((event) {
+          if (event[1] != null) {
+            _controller.add(event[1]);
+          } else {
+            _controller.close();
+          }
+        });
+    return _controller.stream;
+  }
+
+  void _pathWatchedEnd() {
+    _subscription.cancel();
+    _controller.close();
+  }
 }
 
+
 Uint8List _makeUint8ListView(Uint8List source, int offsetInBytes, int length) {
   return new Uint8List.view(source.buffer, offsetInBytes, length);
 }