Add Event delegation to Elements and groups of elements.

tools/dom/src/EventStreamProvider.dart

Index: tools/dom/src/EventStreamProvider.dart
diff --git a/tools/dom/src/EventStreamProvider.dart b/tools/dom/src/EventStreamProvider.dart
index 8e3b0bce8f20cfc6a9ab1d7fb1deb4839cee82b3..2c2a17a18b7127cd9de24288c7dcd5d414c9bc86 100644
--- a/tools/dom/src/EventStreamProvider.dart
+++ b/tools/dom/src/EventStreamProvider.dart
@@ -5,6 +5,75 @@
 part of html;
 
 /**
+ * A pool of streams whose events are unified and emitted through a central
+ * stream.
+ */
+// TODO (efortuna): Remove this when Issue 12218 is addressed.
+class StreamPool<T> {

[blois, 2013/08/08 00:12:11]

Should not be public, can we move it down in the file as well?

[Emily Fortuna, 2013/08/09 00:22:06]

Done.

+  StreamController<T> _controller;

[blois, 2013/08/08 00:12:11]

final?

[Emily Fortuna, 2013/08/09 00:22:06]

can't be because _controller has the onCancel method that calls "close", which
is an instance method, and can't be called in the initializer section.

+
+  /// Subscriptions to the streams that make up the pool.
+  var _subscriptions = new Map<Stream<T>, StreamSubscription<T>>();
+
+  /**
+   * Creates a new stream pool that only supports a single subscriber.
+   *
+   * Any events from broadcast streams in the pool will be buffered until a
+   * listener is subscribed.
+   */
+  StreamPool() {
+    _controller = new StreamController<T>(sync: true, onCancel: close);
+  }
+
+  /**
+   * Creates a new stream pool where [stream] can be listened to more than
+   * once.
+   *
+   * Any events from buffered streams in the pool will be emitted immediately,
+   * regardless of whether [stream] has any subscribers.
+   */
+  StreamPool.broadcast() {
+    _controller = new StreamController<T>.broadcast(sync: true,
+        onCancel: close);
+  }
+
+  /**
+   * The stream through which all events from streams in the pool are emitted.
+   */
+  Stream<T> get stream => _controller.stream;
+
+  /**
+   * Adds [stream] as a member of this pool.
+   *
+   * Any events from [stream] will be emitted through [this.stream]. If
+   * [stream] is sync, they'll be emitted synchronously; if [stream] is async,
+   * they'll be emitted asynchronously.
+   */
+  void add(Stream<T> stream) {
+    if (_subscriptions.containsKey(stream)) return;
+    _subscriptions[stream] = stream.listen(_controller.add,
+        onError: _controller.addError,
+        onDone: () => remove(stream));
+  }
+
+  /** Removes [stream] as a member of this pool. */
+  void remove(Stream<T> stream) {
+    var subscription = _subscriptions.remove(stream);
+    if (subscription != null) subscription.cancel();
+  }
+
+  /** Removes all streams from this pool and closes [stream]. */
+  void close() {
+    for (var subscription in _subscriptions.values) {
+      subscription.cancel();
+    }
+    _subscriptions.clear();
+    _controller.close();
+  }
+}
+
+
+/**
  * Adapter for exposing DOM events as Dart streams.
  */
 class _EventStream<T extends Event> extends Stream<T> {
@@ -30,6 +99,111 @@ class _EventStream<T extends Event> extends Stream<T> {
   }
 }
 
+/** A specialized Stream available to [Element]s to enable event delegation. */
+abstract class ElementStream<T extends Event> extends _EventStream<T> {
+  ElementStream(target, eventType, useCapture) :
+      super(target, eventType, useCapture);
+
+  /**
+   * Return a stream that only fires when the particular event fires for
+   * elements matching the specified CSS selector.
+   *
+   * This is the Dart equivalent to jQuery's
+   * [delegate](http://api.jquery.com/delegate/).
+   */
+  Stream<T> matches(String selector);
+}
+
+/**
+ * Adapter for exposing DOM Element events as streams, while also allowing
+ * event delegation.
+ */
+class _ElementEventStreamImpl<T extends Event> extends ElementStream<T> {
+  _ElementEventStreamImpl(target, eventType, useCapture) :
+      super(target, eventType, useCapture);
+
+  Stream<T> matches(String selector) =>
+      this.where((event) => event.target.matches(selector, true));
+}
+
+/**
+ * Adapter for exposing events on a collection of DOM Elements as streams,
+ * while also allowing event delegation.
+ */
+class _ElementListEventStreamImpl<T extends Event>  implements ElementStream<T> {

[blois, 2013/08/08 00:12:11]

I'm still unclear why this cannot extend Stream and implement ElementStream,
removing the need to forward all of these APIs.

[Emily Fortuna, 2013/08/09 00:22:06]

How would this extend Stream and not need to call all of the forwarding methods?
If you extended Stream, then you're saying that _ElementListEventStreamImpl IS a
stream. But it's not. It masquerades as a stream holding a StreamController/Pool
that produces a stream, but _ElementListEventStreamImpl itself is not a stream.
If I didn't do all the method forwarding, if you called map [taken from
stream.dart], the following code would execute: return new _MapStream<T,
dynamic>(this, convert); but in the case of _ElementListEventStreamImpl, "this"
is not a Stream. Instead you want to call map on _ElementListEventStreamImpl's
stream. 

+//implements Stream<T> {

[blois, 2013/08/08 00:12:11]

Comment

[Emily Fortuna, 2013/08/09 00:22:06]

Done.

+
+  StreamPool _pool;

[blois, 2013/08/08 00:12:11]

final.

[Emily Fortuna, 2013/08/09 00:22:06]

Done.

+  Stream<T> _stream;

[blois, 2013/08/08 00:12:11]

Is it necessary to cache this?

[Emily Fortuna, 2013/08/09 00:22:06]

It's helpful, because otherwise get stream in StreamController creates a new
stream every time you access it.

+
+  _ElementListEventStreamImpl(targetList, eventType, useCapture) {
+    _pool = new StreamPool.broadcast();
+    for (Element target in targetList) {
+      var stream = new _EventStream(target, eventType, useCapture);
+      _pool.add(stream);
+    }
+    _stream = _pool.stream;
+  }
+
+  Stream<T> matches(String selector) =>
+      this.where((event) => event.target.matches(selector, true));
+
+  // Delegate all regular Stream behavor to our wrapped Stream.
+  StreamSubscription<T> listen(void onData(T event),
+      { void onError(error),
+        void onDone(),
+        bool cancelOnError}) =>
+      _stream.listen(onData, onError: onError, onDone: onDone,
+          cancelOnError: cancelOnError);
+  Stream<T> asBroadcastStream({void onListen(StreamSubscription subscription),
+                               void onCancel(StreamSubscription subscription)})
+      => _stream;
+  bool get isBroadcast => true;
+  Stream<T> where(bool test(T event)) => _stream.where(test);
+  Stream map(convert(T event)) => _stream.map(convert);
+  Stream<T> handleError(void handle( error), { bool test(error) }) =>
+      _stream.handleError(handle, test: test);
+  Stream expand(Iterable convert(T value)) =>
+      _stream.expand(convert);
+  Future pipe(StreamConsumer<T> streamConsumer) =>
+      _stream.pipe(streamConsumer);
+  Stream transform(StreamTransformer<T, dynamic> streamTransformer) =>
+      _stream.transform(streamTransformer);
+  Future<T> reduce(T combine(T previous, T element)) =>
+      _stream.reduce(combine);
+  Future fold(var initialValue, combine(var previous, T element)) =>
+      _stream.fold(initialValue, combine);
+  Future<String> join([String separator = ""]) =>
+      _stream.join(separator);
+  Future<bool> contains(Object needle) => _stream.contains(needle);
+  Future forEach(void action(T element)) => _stream.forEach(action);
+  Future<bool> every(bool test(T element)) => _stream.every(test);
+  Future<bool> any(bool test(T element)) => _stream.any(test);
+  Future<int> get length => _stream.length;
+  Future<bool> get isEmpty => _stream.isEmpty;
+  Future<List<T>> toList() => _stream.toList();
+  Future<Set<T>> toSet() => _stream.toSet();
+  Future drain([var futureValue]) => _stream.drain(futureValue);
+  Stream<T> take(int count) => _stream.take(count);
+  Stream<T> takeWhile(bool test(T element)) =>
+      _stream.takeWhile(test);
+  Stream<T> skip(int count) => _stream.skip(count);
+  Stream<T> skipWhile(bool test(T element)) =>
+      _stream.skipWhile(test);
+  Stream<T> distinct([bool equals(T previous, T next)]) =>
+      _stream.distinct(equals);
+  Future<T> get first => _stream.first;
+  Future<T> get last => _stream.last;
+  Future<T> get single => _stream.single;
+  Future<dynamic> firstWhere(bool test(T element), {Object defaultValue()}) =>
+      _stream.firstWhere(test, defaultValue: defaultValue);
+  Future<dynamic> lastWhere(bool test(T element), {Object defaultValue()}) =>
+      _stream.lastWhere(test, defaultValue: defaultValue);
+  Future<T> singleWhere(bool test(T element)) =>
+      _stream.singleWhere(test);
+  Future<T> elementAt(int index) => _stream.elementAt(index);
+}
+
 class _EventStreamSubscription<T extends Event> extends StreamSubscription<T> {
   int _pauseCount = 0;
   EventTarget _target;
@@ -107,7 +281,6 @@ class _EventStreamSubscription<T extends Event> extends StreamSubscription<T> {
   }
 }
 
-
 /**
  * A factory to expose DOM events as Streams.
  */
@@ -135,7 +308,48 @@ class EventStreamProvider<T extends Event> {
    * [addEventListener](http://docs.webplatform.org/wiki/dom/methods/addEventListener)
    */
   Stream<T> forTarget(EventTarget e, {bool useCapture: false}) {
-    return new _EventStream(e, _eventType, useCapture);
+    if (e is Element) {
+      return new _ElementEventStreamImpl(e, _eventType, useCapture);

[blois, 2013/08/08 00:12:11]

Does this need to be here given that there's already forElementTarget?

[Emily Fortuna, 2013/08/09 00:22:06]

I want Elements to always have the ability to do event delegation, even if some
user calls this API, which they can since it's public.... but I changed it as
requested.

+    } else {
+      return new _EventStream(e, _eventType, useCapture);
+    }
+  }
+
+  /**
+   * Gets an [ElementEventStream] for this event type, on the specified element.
+   *
+   * This will always return a broadcast stream so multiple listeners can be
+   * used simultaneously.
+   *
+   * This may be used to capture DOM events:
+   *
+   *     Element.keyDownEvent.forElementTarget(element, useCapture: true).listen(...);
+   *
+   * See also:
+   *
+   * [addEventListener](http://docs.webplatform.org/wiki/dom/methods/addEventListener)
+   */
+  ElementStream<T> forElementTarget(Element e, {bool useCapture: false}) {

[blois, 2013/08/08 00:12:11]

How about just 'forElement'?

+    return new _ElementEventStreamImpl(e, _eventType, useCapture);
+  }
+
+  /**
+   * Gets an [ElementEventStream] for this event type, on the list of elements.
+   *
+   * This will always return a broadcast stream so multiple listeners can be
+   * used simultaneously.
+   *
+   * This may be used to capture DOM events:
+   *
+   *     Element.keyDownEvent._forElementTargetList(element, useCapture: true).listen(...);
+   *
+   * See also:
+   *
+   * [addEventListener](http://docs.webplatform.org/wiki/dom/methods/addEventListener)
+   */
+  ElementStream<T> _forElementTargetList(ElementList e,
+      {bool useCapture: false}) {
+    return new _ElementListEventStreamImpl(e, _eventType, useCapture);
   }
 
   /**
@@ -164,6 +378,15 @@ class _CustomEventStreamProvider<T extends Event>
     return new _EventStream(e, _eventTypeGetter(e), useCapture);
   }
 
+  ElementStream<T> forElementTarget(Element e, {bool useCapture: false}) {
+    return new _ElementEventStreamImpl(e, _eventTypeGetter(e), useCapture);
+  }
+
+  ElementStream<T> _forElementTargetList(ElementList e,
+      {bool useCapture: false}) {
+    return new _ElementListEventStreamImpl(e, _eventTypeGetter(e), useCapture);
+  }
+
   String getEventType(EventTarget target) {
     return _eventTypeGetter(target);
   }