Adding streams to dart:html.

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
new file mode 100644
index 0000000000000000000000000000000000000000..c4e845b2788d7c33ef55c6ede1eddde654e3423e
--- /dev/null
+++ b/tools/dom/src/EventStreamProvider.dart
@@ -0,0 +1,144 @@
+// 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.
+
+part of html;
+
+/**
+ * Adapter for exposing DOM events as Dart streams.
+ */
+class _EventStream<T extends Event> extends Stream<T> {
+  final EventTarget _target;
+  final String _eventType;
+  final bool _useCapture;
+
+  _EventStream(this._target, this._eventType, this._useCapture);
+
+  StreamSubscription<T> listen(void onData(T event),
+      { void onError(AsyncError error),
+      void onDone(),
+      bool unsubscribeOnError}) {
+
+    if (onError != null) {
+      throw new ArgumentError('onError is not supported by DOM events');

[floitsch, 2013/01/12 00:49:33]

I would just discard it. It is supported, but the stream will just never emit
it.

[blois, 2013/01/12 01:42:52]

Done.

Old compatibility habits- always throw exceptions in case support is added
later, otherwise it'd be a breaking change (.NET rules at least!)

+    }
+    if (onDone != null) {

[floitsch, 2013/01/12 00:49:33]

ditto.

[blois, 2013/01/12 01:42:52]

Done.

+      throw new ArgumentError('onError is not supported by DOM events');
+    }
+    if (unsubscribeOnError != null) {

[floitsch, 2013/01/12 00:49:33]

ditto.

[blois, 2013/01/12 01:42:52]

Done.

+      throw new ArgumentError(
+          'unsubscribeOnError is not supported by DOM events');
+    }
+
+    return new _EventStreamSubscription<T>(
+        this._target, this._eventType, onData, this._useCapture);
+  }
+}
+
+class _EventStreamSubscription<T extends Event> extends StreamSubscription<T> {

[floitsch, 2013/01/12 00:49:33]

This is pretty much what I was looking for.

I don't remember all contracts by heart (so I'm not sure if you are allowed to
resume a non-paused stream...), but otherwise LGTM!

[Jennifer Messerly, 2013/01/12 01:11:25]

+1, this is so nice

+  int _pauseCount = 0;
+  EventTarget _target;
+  final String _eventType;
+  var _onData;
+  final bool _useCapture;
+
+  _EventStreamSubscription(this._target, this._eventType, this._onData,
+      this._useCapture) {
+    _tryResume();
+  }
+
+  void cancel() {
+    if (_canceled) {
+      throw new StateError("Subscription has been canceled.");
+    }
+
+    _unlisten();
+    // Clear out the target to indicate this is complete.

[Jennifer Messerly, 2013/01/12 01:11:25]

also has the nice side effect of releasing memory :)
might want to set _onData to null too, either here or in _unlisten

[blois, 2013/01/12 01:42:52]

Done.

+    _target = null;
+  }
+
+  bool get _canceled => _target == null;
+
+  void onData(void handleData(T event)) {
+    if (_canceled) {
+      throw new StateError("Subscription has been canceled.");
+    }
+    // Remove current event listener.
+    _unlisten();
+
+    _onData = handleData;

[Jennifer Messerly, 2013/01/12 01:11:25]

is this allowed to be null? I see some checks for it, but wasn't sure how it can
happen

[blois, 2013/01/12 01:42:52]

onData can be null from the listen call and I believe it can be set to null at
any point. I added a test for these cases.

+    _tryResume();
+  }
+
+  void onError(void handleError(AsyncError error)) {
+    throw new UnsupportedError('Not supported by DOM events');

[floitsch, 2013/01/12 00:49:33]

ditto: just ignore it.

[blois, 2013/01/12 01:42:52]

Done.

+  }
+
+  void onDone(void handleDone()) {
+    throw new UnsupportedError('Not supported by DOM events');
+  }
+
+  void pause([Future resumeSignal]) {
+    if (_canceled) {
+      throw new StateError("Subscription has been canceled.");
+    }
+    ++_pauseCount;
+    _unlisten();
+
+    if (resumeSignal != null) {
+      resumeSignal.whenComplete(resume);
+    }
+  }
+
+  void resume() {

[Jennifer Messerly, 2013/01/12 01:11:25]

this should probably check if paused. The _StreamSubscriptionImpl does this:

    if (!isPaused) {
      throw new StateError("Resuming unpaused subscription");
    }

[blois, 2013/01/12 01:42:52]

Done.

+    if (_canceled) {
+      throw new StateError("Subscription has been canceled.");
+    }
+    if (_pauseCount == 0) {
+      throw new StateError("Subscription is not paused.");
+    }
+    --_pauseCount;
+    _tryResume();
+  }
+
+  void _tryResume() {
+    if (_onData != null && _pauseCount == 0) {
+      _target.$dom_addEventListener(_eventType, _onData, _useCapture);
+    }
+  }
+
+  void _unlisten() {
+    if (_onData != null) {
+      _target.$dom_removeEventListener(_eventType, _onData, _useCapture);

[Jennifer Messerly, 2013/01/12 01:11:25]

should this set _onData to null after unregistering?

[blois, 2013/01/12 01:42:52]

We unlisten when it's paused, but can resume later. I added clearing _onData in
cancel since that's not recoverable.

+    }
+  }
+}
+
+
+/**
+ * A factory to expose DOM events as Streams.
+ */
+class EventStreamProvider<T extends Event> {
+  final String _eventType;
+
+  const EventStreamProvider(this._eventType);
+
+  /**
+   * Gets a [Stream] for this event type, on the specified target.
+   *
+   * This may be used to capture DOM events:
+   *
+   *     Element.keyDownEvent.forTarget(element, useCapture: true).listen(...);
+   *
+   * Or for listening to an event which will bubble through the DOM tree:
+   *
+   *     MediaElement.pauseEvent.forTarget(document.body).listen(...);
+   *
+   * See also:
+   *
+   * [addEventListener](http://docs.webplatform.org/wiki/dom/methods/addEventListener)
+   */
+  Stream<T> forTarget(EventTarget e, {bool useCapture: false}) {
+    return new _EventStream(e, _eventType, useCapture);
+  }
+}