big update to observe, template_binding, polymer

pkg/observe/lib/src/path_observer.dart

Index: pkg/observe/lib/src/path_observer.dart
diff --git a/pkg/observe/lib/src/path_observer.dart b/pkg/observe/lib/src/path_observer.dart
index 5e59b80d1f592f1321a56d12e54686ca5478280b..8054a3ab19d38cac680f14853aa7c197a445337a 100644
--- a/pkg/observe/lib/src/path_observer.dart
+++ b/pkg/observe/lib/src/path_observer.dart
@@ -5,6 +5,8 @@
 library observe.src.path_observer;
 
 import 'dart:async';
+import 'dart:collection';
+import 'dart:math' show min;
 @MirrorsUsed(metaTargets: const [Reflectable, ObservableProperty],
     override: 'observe.src.path_observer')
 import 'dart:mirrors';
@@ -12,209 +14,216 @@ import 'package:logging/logging.dart' show Logger, Level;
 import 'package:observe/observe.dart';
 import 'package:observe/src/observable.dart' show objectType;
 
-// This code is inspired by ChangeSummary:
-// https://github.com/rafaelw/ChangeSummary/blob/master/change_summary.js
-// ...which underlies MDV. Since we don't need the functionality of
-// ChangeSummary, we just implement what we need for data bindings.
-// This allows our implementation to be much simpler.
-
-/**
- * A data-bound path starting from a view-model or model object, for example
- * `foo.bar.baz`.
- *
- * When the [values] stream is being listened to, this will observe changes to
- * the object and any intermediate object along the path, and send [values]
- * accordingly. When all listeners are unregistered it will stop observing
- * the objects.
- *
- * This class is used to implement [Node.bind] and similar functionality.
- */
+/// A data-bound path starting from a view-model or model object, for example
+/// `foo.bar.baz`.
+///
+/// When [open] is called, this will observe changes to the object and any
+/// intermediate object along the path, and send updated values accordingly.
+/// When [close] is called it will stop observing the objects.
+///
+/// This class is used to implement `Node.bind` and similar functionality in
+/// the [template_binding](pub.dartlang.org/packages/template_binding) package.
+class PathObserver extends _Observer implements Bindable {
+  PropertyPath _path;
+  Object _object;
+  _ObservedSet _directObserver;
+
+  /// Observes [path] on [object] for changes. This returns an object
+  /// that can be used to get the changes and get/set the value at this path.
+  ///
+  /// The path can be a [PropertyPath], or a [String] used to construct it.
+  ///
+  /// See [open] and [value].
+  PathObserver(Object object, [path])
+      : _object = object,
+        _path = path is PropertyPath ? path : new PropertyPath(path);
+
+  bool get _isClosed => _path == null;
+
+  /// Sets the value at this path.
+  @reflectable void set value(Object newValue) {
+    if (_path != null) _path.setValueFrom(_object, newValue);
+  }
+
+  int get _reportArgumentCount => 2;
+
+  /// Initiates observation and returns the initial value.
+  /// The callback will be passed the updated [value], and may optionally be
+  /// declared to take a second argument, which will contain the previous value.
+  open(callback) => super.open(callback);
+
+  void _connect() {
+    _directObserver = new _ObservedSet(this, _object);
+    _check(skipChanges: true);
+  }
+
+  void _disconnect() {
+    _value = null;
+    if (_directObserver != null) {
+      _directObserver.close(this);
+      _directObserver = null;
+    }
+    // Dart note: the JS impl does not do this, but it seems consistent with
+    // CompoundObserver. After closing the PathObserver can't be reopened.
+    _path = null;
+    _object = null;
+  }
+
+  void _iterateObjects(void observe(obj)) {
+    _path._iterateObjects(_object, observe);
+  }
+
+  bool _check({bool skipChanges: false}) {
+    var oldValue = _value;
+    _value = _path.getValueFrom(_object);
+    if (skipChanges || _value == oldValue) return false;
+
+    _report(_value, oldValue);
+    return true;
+  }
+}
+
+/// A dot-delimieted property path such as "foo.bar" or "foo.10.bar".
+/// The path specifies how to get a particular value from an object graph, where
+/// the graph can include arrays.
 // TODO(jmesserly): consider specialized subclasses for:
 // * empty path
 // * "value"
 // * single token in path, e.g. "foo"
-class PathObserver extends ChangeNotifier {
-  /** The path string. */
-  final String path;
-
-  /** True if the path is valid, otherwise false. */
-  final bool _isValid;
-
+class PropertyPath {
+  /// The segments of the path.
   final List<Object> _segments;
-  List<Object> _values;
-  List<StreamSubscription> _subs;
-
-  final Function _computeValue;
-
-  /**
-   * Observes [path] on [object] for changes. This returns an object that can be
-   * used to get the changes and get/set the value at this path.
-   *
-   * You can optionally use [computeValue] to apply a function to the result of
-   * evaluating the path. The function should be pure, as PathObserver will not
-   * know to observe any of its dependencies. If you need to observe mutliple
-   * values, use [CompoundPathObserver] instead.
-   *
-   * See [PathObserver.bindSync] and [PathObserver.value].
-   */
-  PathObserver(Object object, String path, {computeValue(newValue)})
-      : path = path,
-        _computeValue = computeValue,
-        _isValid = _isPathValid(path),
-        _segments = <Object>[] {
-
-    if (_isValid) {
-      for (var segment in path.trim().split('.')) {
-        if (segment == '') continue;
-        var index = int.parse(segment, radix: 10, onError: (_) => null);
-        _segments.add(index != null ? index : new Symbol(segment));
+
+  /// Creates a new [PropertyPath]. These can be stored to avoid excessive
+  /// parsing of path strings.
+  ///
+  /// The provided [path] should be a String or a List. If it is a list it
+  /// should contain only Symbols and integers. This can be used to avoid
+  /// parsing.
+  ///
+  /// Note that this constructor will canonicalize identical paths in some cases
+  /// to save memory, but this is not guaranteed. Use [==] for comparions
+  /// purposes instead of [identical].
+  factory PropertyPath([path]) {
+    if (path is List) {
+      var copy = new List.from(path, growable: false);
+      for (var segment in copy) {
+        if (segment is! int && segment is! Symbol) {
+          throw new ArgumentError('List must contain only ints and Symbols');
+        }
       }
+      return new PropertyPath._(copy);
     }
 
-    // Initialize arrays.
-    // Note that the path itself can't change after it is initially
-    // constructed, even though the objects along the path can change.
-    _values = new List<Object>(_segments.length + 1);
+    if (path == null) path = '';
 
-    // If we have an empty path, we need to apply the transformation function
-    // to the value. The "value" property should always show the transformed
-    // value.
-    if (_segments.isEmpty && computeValue != null) {
-      object = computeValue(object);
-    }
+    var pathObj = _pathCache[path];
+    if (pathObj != null) return pathObj;
 
-    _values[0] = object;
-    _subs = new List<StreamSubscription>(_segments.length);
-  }
+    if (!_isPathValid(path)) return _InvalidPropertyPath._instance;
 
-  /** The object being observed. If the path is empty this will be [value]. */
-  get object => _values[0];
+    final segments = [];
+    for (var segment in path.trim().split('.')) {
+      if (segment == '') continue;
+      var index = int.parse(segment, radix: 10, onError: (_) => null);
+      segments.add(index != null ? index : new Symbol(segment));
+    }
 
-  /** Gets the last reported value at this path. */
-  @reflectable get value {
-    if (!_isValid) return null;
-    if (!hasObservers) _updateValues();
-    return _values.last;
+    // TODO(jmesserly): we could use an UnmodifiableListView here, but that adds
+    // memory overhead.
+    pathObj = new PropertyPath._(segments.toList(growable: false));
+    if (_pathCache.length >= _pathCacheLimit) {
+      _pathCache.remove(_pathCache.keys.first);
+    }
+    _pathCache[path] = pathObj;
+    return pathObj;
   }
 
-  /** Sets the value at this path. */
-  @reflectable void set value(Object newValue) {
-    int len = _segments.length;
+  PropertyPath._(this._segments);
 
-    // TODO(jmesserly): throw if property cannot be set?
-    // MDV seems tolerant of these errors.
-    if (len == 0) return;
-    if (!hasObservers) _updateValues(end: len - 1);
+  int get length => _segments.length;
+  bool get isEmpty => _segments.isEmpty;
+  bool get isValid => true;
 
-    if (_setObjectProperty(_values[len - 1], _segments[len - 1], newValue)) {
-      // Technically, this would get updated asynchronously via a change record.
-      // However, it is nice if calling the getter will yield the same value
-      // that was just set. So we use this opportunity to update our cache.
-      _values[len] = newValue;
-    }
+  String toString() {
+    if (!isValid) return '<invalid path>';
+    return _segments
+        .map((s) => s is Symbol ? MirrorSystem.getName(s) : s)
+        .join('.');
   }
 
-  /**
-   * Invokes the [callback] immediately with the current [value], and every time
-   * the value changes. This is useful for bindings, which want to be up-to-date
-   * immediately and stay bound to the value of the path.
-   */
-  StreamSubscription bindSync(void callback(value)) {
-    var result = changes.listen((records) { callback(value); });
-    callback(value);
-    return result;
-  }
+  bool operator ==(other) {
+    if (identical(this, other)) return true;
+    if (other is! PropertyPath) return false;
+    if (isValid != other.isValid) return false;
 
-  void observed() {
-    super.observed();
-    _updateValues();
-    _observePath();
+    int len = _segments.length;
+    if (len != other._segments.length) return false;
+    for (int i = 0; i < len; i++) {
+      if (_segments[i] != other._segments[i]) return false;
+    }
+    return true;
   }
 
-  void unobserved() {
-    for (int i = 0; i < _subs.length; i++) {
-      if (_subs[i] != null) {
-        _subs[i].cancel();
-        _subs[i] = null;
-      }
+  /// This is the [Jenkins hash function][1] but using masking to keep
+  /// values in SMI range.
+  /// [1]: http://en.wikipedia.org/wiki/Jenkins_hash_function
+  // TODO(jmesserly): should reuse this instead, see
+  // https://code.google.com/p/dart/issues/detail?id=11617
+  int get hashCode {
+    int hash = 0;
+    for (int i = 0, len = _segments.length; i < len; i++) {
+      hash = 0x1fffffff & (hash + _segments[i].hashCode);
+      hash = 0x1fffffff & (hash + ((0x0007ffff & hash) << 10));
+      hash = hash ^ (hash >> 6);
     }
-    super.unobserved();
+    hash = 0x1fffffff & (hash + ((0x03ffffff & hash) <<  3));
+    hash = hash ^ (hash >> 11);
+    return 0x1fffffff & (hash + ((0x00003fff & hash) << 15));
   }
 
-  // TODO(jmesserly): should we be caching these values if not observing?
-  void _updateValues({int end}) {
-    if (end == null) end = _segments.length;
-    int last = _segments.length - 1;
-    for (int i = 0; i < end; i++) {
-      var newValue = _getObjectProperty(_values[i], _segments[i]);
-      if (i == last && _computeValue != null) {
-        newValue = _computeValue(newValue);
-      }
-      _values[i + 1] = newValue;
+  /// Returns the current of the path from the provided [obj]ect.
+  getValueFrom(Object obj) {
+    if (!isValid) return null;
+    for (var segment in _segments) {
+      if (obj == null) return null;
+      obj = _getObjectProperty(obj, segment);
     }
+    return obj;
   }
 
-  void _updateObservedValues({int start: 0}) {
-    var oldValue, newValue;
-    for (int i = start, last = _segments.length - 1; i <= last; i++) {
-      oldValue = _values[i + 1];
-      newValue = _getObjectProperty(_values[i], _segments[i]);
-      if (i == last && _computeValue != null) {
-        newValue = _computeValue(newValue);
-      }
-      if (identical(oldValue, newValue)) {
-        _observePath(start, i);
-        return;
-      }
-      _values[i + 1] = newValue;
+  /// Attempts to set the [value] of the path from the provided [obj]ect.
+  /// Returns true if and only if the path was reachable and set.
+  bool setValueFrom(Object obj, Object value) {
+    var end = _segments.length - 1;
+    if (end < 0) return false;
+    for (int i = 0; i < end; i++) {
+      if (obj == null) return false;
+      obj = _getObjectProperty(obj, _segments[i]);
     }
-
-    _observePath(start);
-    notifyPropertyChange(#value, oldValue, newValue);
+    return _setObjectProperty(obj, _segments[end], value);
   }
 
-  void _observePath([int start = 0, int end]) {
-    if (end == null) end = _segments.length;
+  void _iterateObjects(Object obj, void observe(obj)) {
+    if (!isValid || isEmpty) return;
 
-    for (int i = start; i < end; i++) {
-      if (_subs[i] != null) _subs[i].cancel();
-      _observeIndex(i);
-    }
-  }
+    int i = 0, last = _segments.length - 1;
+    while (obj != null) {
+      observe(obj);
 
-  void _observeIndex(int i) {
-    final object = _values[i];
-    final segment = _segments[i];
-    if (segment is int) {
-      if (object is ObservableList) {
-        _subs[i] = object.listChanges.listen((List<ListChangeRecord> records) {
-          for (var record in records) {
-            if (record.indexChanged(segment)) {
-              _updateObservedValues(start: i);
-              return;
-            }
-          }
-        });
-      }
-    } else if (object is Observable) {
-      // TODO(jmesserly): rather than allocating a new closure for each
-      // property, we could try and have one for the entire path. However we'd
-      // need to do a linear scan to find the index as soon as we got a change.
-      // Also we need to fix ListChangeRecord and MapChangeRecord to contain
-      // the target. Not sure if it's worth it.
-
-      _subs[i] = object.changes.listen((List<ChangeRecord> records) {
-        for (var record in records) {
-          if (_changeRecordMatches(record, segment)) {
-            _updateObservedValues(start: i);
-            return;
-          }
-        }
-      });
+      if (i >= last) break;
+      obj = _getObjectProperty(obj, _segments[i++]);
     }
   }
 }
 
+class _InvalidPropertyPath extends PropertyPath {
+  static final _instance = new _InvalidPropertyPath();
+
+  bool get isValid => false;
+  _InvalidPropertyPath() : super._([]);
+}
+
 bool _changeRecordMatches(record, key) {
   if (record is PropertyChangeRecord) {
     return (record as PropertyChangeRecord).name == key;
@@ -318,10 +327,8 @@ bool _maybeHasSetter(ClassMirror type, Symbol name) {
   return false;
 }
 
-/**
- * True if the type has a method, other than on Object.
- * Doesn't consider noSuchMethod, unless [name] is `#noSuchMethod`.
- */
+/// True if the type has a method, other than on Object.
+/// Doesn't consider noSuchMethod, unless [name] is `#noSuchMethod`.
 bool _hasMethod(ClassMirror type, Symbol name) {
   while (type != objectType) {
     final member = type.declarations[name];
@@ -356,14 +363,376 @@ final _pathRegExp = () {
   return new RegExp('^$path\$');
 }();
 
-final _spacesRegExp = new RegExp(r'\s');
-
 bool _isPathValid(String s) {
-  s = s.replaceAll(_spacesRegExp, '');
-
+  s = s.trim();
   if (s == '') return true;
   if (s[0] == '.') return false;
   return _pathRegExp.hasMatch(s);
 }
 
 final Logger _logger = new Logger('observe.PathObserver');
+
+
+/// This is a simple cache. It's like LRU but we don't update an item on a
+/// cache hit, because that would require allocation. Better to let it expire
+/// and reallocate the PropertyPath.
+// TODO(jmesserly): this optimization is from observe-js, how valuable is it in
+// practice?
+final _pathCache = new LinkedHashMap<String, PropertyPath>();
+
+/// The size of a path like "foo.bar" is approximately 160 bytes, so this
+/// reserves ~16Kb of memory for recently used paths. Since paths are frequently
+/// reused, the theory is that this ends up being a good tradeoff in practice.
+// (Note: the 160 byte estimate is from Dart VM 1.0.0.10_r30798 on x64 without
+// using UnmodifiableListView in PropertyPath)
+const int _pathCacheLimit = 100;
+
+/// [CompoundObserver] is a [Bindable] object which knows how to listen to
+/// multiple values (registered via [addPath] or [addObserver]) and invoke a
+/// callback when one or more of the values have changed.
+///
+///    var obj = new ObservableMap.from({'a': 1, 'b': 2});
+///    var otherObj = new ObservableMap.from({'c': 3});
+///
+///    var observer = new CompoundObserver()
+///      ..addPath(obj, 'a');
+///      ..addObserver(new PathObserver(obj, 'b'));
+///      ..addPath(otherObj, 'c');
+///      ..open((values) {
+///        for (int i = 0; i < values.length; i++) {
+///          print('The value at index $i is now ${values[i]}');
+///        }
+///      });
+///
+///   obj['a'] = 10; // print will be triggered async
+///
+class CompoundObserver extends _Observer implements Bindable {
+  _ObservedSet _directObserver;
+  List _observed = [];
+
+  bool get _isClosed => _observed == null;
+
+  CompoundObserver() {
+    _value = [];
+  }
+
+  int get _reportArgumentCount => 3;
+
+  /// Initiates observation and returns the initial value.
+  /// The callback will be passed the updated [value], and may optionally be
+  /// declared to take a second argument, which will contain the previous value.
+  ///
+  /// Implementation note: a third argument can also be declared, which will
+  /// receive a list of objects and paths, such that `list[2 * i]` will access
+  /// the object and `list[2 * i + 1]` will access the path, where `i` is the
+  /// order of the [addPath] call. This parameter is only used by
+  /// `package:polymer` as a performance optimization, and should not be relied
+  /// on in new code.
+  open(callback) => super.open(callback);
+
+  void _connect() {
+    _check(skipChanges: true);
+
+    for (var i = 0; i < _observed.length; i += 2) {
+      var object = _observed[i];
+      if (!identical(object, _observerSentinel)) {
+        _directObserver = new _ObservedSet(this, object);
+        break;
+      }
+    }
+  }
+
+  void _disconnect() {
+    _value = null;
+
+    if (_directObserver != null) {
+      _directObserver.close(this);
+      _directObserver = null;
+    }
+
+    for (var i = 0; i < _observed.length; i += 2) {
+      if (identical(_observed[i], _observerSentinel)) {
+        _observed[i + 1].close();
+      }
+    }
+    _observed = null;
+  }
+
+  /// Adds a dependency on the property [path] accessed from [object].
+  /// [path] can be a [PropertyPath] or a [String]. If it is omitted an empty
+  /// path will be used.
+  void addPath(Object object, [path]) {
+    if (_isOpen || _isClosed) {
+      throw new StateError('Cannot add paths once started.');
+    }
+
+    if (path is! PropertyPath) path = new PropertyPath(path);
+    _observed..add(object)..add(path);
+  }
+
+  void addObserver(Bindable observer) {
+    if (_isOpen || _isClosed) {
+      throw new StateError('Cannot add observers once started.');
+    }
+
+    observer.open(_deliver);
+    _observed..add(_observerSentinel)..add(observer);
+  }
+
+  void _iterateObjects(void observe(obj)) {
+    for (var i = 0; i < _observed.length; i += 2) {
+      var object = _observed[i];
+      if (!identical(object, _observerSentinel)) {
+        (_observed[i + 1] as PropertyPath)._iterateObjects(object, observe);
+      }
+    }
+  }
+
+  bool _check({bool skipChanges: false}) {
+    bool changed = false;
+    _value.length = _observed.length ~/ 2;
+    var oldValues = null;
+    for (var i = 0; i < _observed.length; i += 2) {
+      var pathOrObserver = _observed[i + 1];
+      var object = _observed[i];
+      var value = identical(object, _observerSentinel) ?
+          (pathOrObserver as Bindable).value :
+          (pathOrObserver as PropertyPath).getValueFrom(object);
+
+      if (skipChanges) {
+        _value[i ~/ 2] = value;
+        continue;
+      }
+
+      if (value == _value[i ~/ 2]) continue;
+
+      // don't allocate this unless necessary.
+      if (_notifyArgumentCount >= 2) {
+        if (oldValues == null) oldValues = new Map();
+        oldValues[i ~/ 2] = _value[i ~/ 2];
+      }
+
+      changed = true;
+      _value[i ~/ 2] = value;
+    }
+
+    if (!changed) return false;
+
+    // TODO(rafaelw): Having _observed as the third callback arg here is
+    // pretty lame API. Fix.
+    _report(_value, oldValues, _observed);
+    return true;
+  }
+}
+
+const _observerSentinel = const _ObserverSentinel();
+class _ObserverSentinel { const _ObserverSentinel(); }
+
+// A base class for the shared API implemented by PathObserver and
+// CompoundObserver and used in _ObservedSet.
+abstract class _Observer extends Bindable {
+  static int _nextBirthId = 0;
+
+  /// A number indicating when the object was created.
+  final int _birthId = _nextBirthId++;
+
+  Function _notifyCallback;
+  int _notifyArgumentCount;
+  var _value;
+
+  // abstract members
+  void _iterateObjects(void observe(obj));
+  void _connect();
+  void _disconnect();
+  bool get _isClosed;
+  _check({bool skipChanges: false});
+
+  bool get _isOpen => _notifyCallback != null;
+
+  /// The number of arguments the subclass will pass to [_report].
+  int get _reportArgumentCount;
+
+  open(callback) {
+    if (_isOpen || _isClosed) {
+      throw new StateError('Observer has already been opened.');
+    }
+
+    if (_minArgumentCount(callback) > _reportArgumentCount) {
+      throw new ArgumentError('callback should take $_reportArgumentCount or '
+          'fewer arguments');
+    }
+
+    _notifyCallback = callback;
+    _notifyArgumentCount = min(_reportArgumentCount,
+        _maxArgumentCount(callback));
+
+    _connect();
+    return _value;
+  }
+
+  @reflectable get value {
+    _check(skipChanges: true);
+    return _value;
+  }
+
+  void close() {
+    if (!_isOpen) return;
+
+    _disconnect();
+    _value = null;
+    _notifyCallback = null;
+  }
+
+  void _deliver(_) {
+    if (_isOpen) _dirtyCheck();
+  }
+
+  bool _dirtyCheck() {
+    var cycles = 0;
+    while (cycles < _MAX_DIRTY_CHECK_CYCLES && _check()) {
+      cycles++;
+    }
+    return cycles > 0;
+  }
+
+  void _report(newValue, oldValue, [extraArg]) {
+    try {
+      switch (_notifyArgumentCount) {
+        case 0: _notifyCallback(); break;
+        case 1: _notifyCallback(newValue); break;
+        case 2: _notifyCallback(newValue, oldValue); break;
+        case 3: _notifyCallback(newValue, oldValue, extraArg); break;
+      }
+    } catch (e, s) {
+      // Deliver errors async, so if a single callback fails it doesn't prevent
+      // other things from working.
+      new Completer().completeError(e, s);
+    }
+  }
+}
+
+typedef _Func0();
+typedef _Func1(a);
+typedef _Func2(a, b);
+typedef _Func3(a, b, c);
+
+int _minArgumentCount(fn) {
+  if (fn is _Func0) return 0;
+  if (fn is _Func1) return 1;
+  if (fn is _Func2) return 2;
+  if (fn is _Func3) return 3;
+  return 4; // at least 4 arguments are required.
+}
+
+int _maxArgumentCount(fn) {
+  if (fn is _Func3) return 3;
+  if (fn is _Func2) return 2;
+  if (fn is _Func1) return 1;
+  if (fn is _Func0) return 0;
+  return -1;
+}
+
+class _ObservedSet {
+  /// To prevent sequential [PathObserver]s and [CompoundObserver]s from
+  /// observing the same object, we check if they are observing the same root
+  /// as the most recently created observer, and if so merge it into the
+  /// existing _ObservedSet.
+  ///
+  /// See <https://github.com/Polymer/observe-js/commit/f0990b1> and
+  /// <https://codereview.appspot.com/46780044/>.
+  static _ObservedSet _lastSet;
+
+  /// The root object for a [PathObserver]. For a [CompoundObserver], the root
+  /// object of the first path observed. This is used by the constructor to
+  /// reuse an [_ObservedSet] that starts from the same object.
+  Object _rootObject;
+
+  /// Observers associated with this root object, in birth order.
+  final Map<int, _Observer> _observers = new SplayTreeMap();
+
+  // Dart note: the JS implementation is O(N^2) because Array.indexOf is used
+  // for lookup in these two arrays. We use HashMap to avoid this problem. It
+  // also gives us a nice way of tracking the StreamSubscription.
+  Map<Object, StreamSubscription> _objects;
+  Map<Object, StreamSubscription> _toRemove;
+
+  bool _resetNeeded = false;
+
+  factory _ObservedSet(_Observer observer, Object rootObj) {
+    if (_lastSet == null || !identical(_lastSet._rootObject, rootObj)) {
+      _lastSet = new _ObservedSet._(rootObj);
+    }
+    _lastSet.open(observer);
+  }
+
+  _ObservedSet._(this._rootObject);
+
+  void open(_Observer obs) {
+    _observers[obs._birthId] = obs;
+    obs._iterateObjects(observe);
+  }
+
+  void close(_Observer obs) {
+    var anyLeft = false;
+
+    _observers.remove(obs._birthId);
+
+    if (_observers.isNotEmpty) {
+      _resetNeeded = true;
+      scheduleMicrotask(reset);
+      return;
+    }
+    _resetNeeded = false;
+
+    if (_objects != null) {
+      for (var sub in _objects) sub.cancel();
+      _objects = null;
+    }
+  }
+
+  void observe(Object obj) {
+    if (obj is ObservableList) _observeStream(obj.listChanges);
+    if (obj is Observable) _observeStream(obj.changes);
+  }
+
+  void _observeStream(Stream stream) {
+    // TODO(jmesserly): we hash on streams as we have two separate change
+    // streams for ObservableList. Not sure if that is the design we will use
+    // going forward.
+
+    if (_objects == null) _objects = new HashMap();
+    StreamSubscription sub = null;
+    if (_toRemove != null) sub = _toRemove.remove(stream);
+    if (sub != null) {
+      _objects[stream] = sub;
+    } else if (!_objects.containsKey(stream)) {
+      _objects[stream] = stream.listen(_callback);
+    }
+  }
+
+  void reset() {
+    if (!_resetNeeded) return;
+
+    var objs = _toRemove == null ? new HashMap() : _toRemove;
+    _toRemove = _objects;
+    _objects = objs;
+    for (var observer in _observers.values) {
+      if (observer._isOpen) observer._iterateObjects(observe);
+    }
+
+    for (var sub in _toRemove.values) sub.cancel();
+
+    _toRemove = null;
+  }
+
+  void _callback(records) {
+    for (var observer in _observers.values.toList(growable: false)) {
+      if (observer._isOpen) observer._check();
+    }
+
+    _resetNeeded = true;
+    scheduleMicrotask(reset);
+  }
+}
+
+const int _MAX_DIRTY_CHECK_CYCLES = 1000;