Add Stream.timeout method.

sdk/lib/async/stream.dart

Index: sdk/lib/async/stream.dart
diff --git a/sdk/lib/async/stream.dart b/sdk/lib/async/stream.dart
index 58bb090cba99e14547def60c9a5e0d31e43dede9..2734021a0ea0f203db481642ec5dc869e165379c 100644
--- a/sdk/lib/async/stream.dart
+++ b/sdk/lib/async/stream.dart
@@ -649,7 +649,7 @@ abstract class Stream<T> {
    * as [test] returns [:true:] for the event data. The stream is done
    * when either this stream is done, or when this stream first provides
    * a value that [test] doesn't accept.
-   * 
+   *
    * Stops listening to the stream after the accepted elements.
    *
    * Internally the method cancels its subscription after these elements. This
@@ -960,6 +960,84 @@ abstract class Stream<T> {
       cancelOnError: true);
     return future;
   }
+
+  /**
+   * Creates a new stream with the same events as this stream.
+   *
+   * Whenever more than [timeLimit] passes between two events from this stream,
+   * the [onTimeout] function is called.
+   *
+   * The countdown doesn't start until the returned stream is listened to.
+   * The countdown is reset every time an event is forwarded from this stream,
+   * or when the stream is paused and resumed.
+   *
+   * The [onTimeout] function is called with one argument: an
+   * [EventSink] that allows putting events into the returned stream.
+   * This `EventSink` is only valid during the call to `onTimeout`.
+   *
+   * If `onTimeout` is omitted, a timeout will just put a [TimeoutException]
+   * into the error channel of the returned stream.
+   */
+  Stream timeout(Duration timeLimit, {void onTimeout(EventSink sink)}) {
+    StreamSubscription<T> subscription;
+    _StreamController controller;
+    // The following variables are set on listen.
+    Timer timer;
+    Zone zone;
+    Function timeout;
+
+    void onData(T event) {
+      timer.cancel();
+      controller.add(event);
+      timer = zone.createTimer(timeLimit, timeout);
+    }
+    void onError(error, StackTrace stackTrace) {
+      timer.cancel();
+      controller.addError(error, stackTrace);
+      timer = zone.createTimer(timeLimit, timeout);
+    }
+    void onDone() {
+      timer.cancel();
+      controller.close();
+    }
+    controller = new _SyncStreamController(
+        () {
+          zone = Zone.current;

[floitsch, 2013/12/05 12:37:38]

As discussed:
we should just use the subscription's zone, and not the one that invoked the
timeout.

[Lasse Reichstein Nielsen, 2013/12/05 13:28:44]

That is what we are doing now.
The zone is being set to the current zone at the time of listening (this code is
the onListen callback, called right after creating the subscription object).
I didn't take it from the actual subscription, because this code is shorter than
controller._subscription._zone, and makes fewer assumptions about the internal
structure of controller and subscription, and it does the same thing anyway (the
_zone field is simply declared as "final Zone _zone = Zone.current;") 

[floitsch, 2013/12/05 14:20:49]

yes. didn't see that the code was already changed.

+          if (onTimeout == null) {
+            timeout = () {
+              controller.addError(new TimeoutException("No stream event",
+                                                       timeLimit));
+            };
+          } else {
+            onTimeout = zone.registerUnaryCallback(onTimeout);
+            _ControllerEventSinkWrapper wrapper =
+                new _ControllerEventSinkWrapper(null);
+            timeout = () {
+              wrapper._sink = controller;  // Only valid during call.
+              zone.runUnaryGuarded(onTimeout, wrapper);

[floitsch, 2013/12/05 14:20:49]

It's actually not runUnaryGuarded.
If the onTimeout throws, it should be propagated to the listeners and not be
considered an unhandled error.
Please add a test for this, too.

[Lasse Reichstein Nielsen, 2013/12/05 14:57:54]

I can see the point.
I actually preferred to have the onTimeout failure not show up in the stream,
because it should be considered a programming error (you can put errors into the
sink, so throwing isn't the right way to make an error).
On the other hand, it makes it harder to handle the error if you can't catch it
downstream, but need an error-zone for it.

+              wrapper._sink = null;
+            };
+          }
+
+          subscription = this.listen(onData, onError: onError, onDone: onDone);
+          timer = zone.createTimer(timeLimit, timeout);
+        },
+        () {
+          timer.cancel();
+          subscription.pause();
+        },
+        () {
+          subscription.resume();
+          timer = zone.createTimer(timeLimit, timeout);
+        },
+        () {
+          timer.cancel();
+          Future result = subscription.cancel();
+          subscription = null;
+          return result;
+        });
+    return controller.stream;
+  }
 }
 
 /**
@@ -1282,3 +1360,18 @@ abstract class StreamIterator<T> {
    */
   Future cancel();
 }
+
+
+/**
+ * Wraps an [_EventSink] so it exposes only the [EventSink] interface.
+ */
+class _ControllerEventSinkWrapper<T> implements EventSink<T> {
+  EventSink _sink;
+  _ControllerEventSinkWrapper(this._sink);
+
+  void add(T data) { _sink.add(data); }
+  void addError(error, [StackTrace stackTrace]) {
+    _sink.addError(error, stackTrace);
+  }
+  void close() { _sink.close(); }
+}