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..7dd348e93b061d6bb9d35a8623daf32735bbb897 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,101 @@ 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.
+   *
+   * If the [onTimeout] function accepts one argument, it is called with an
+   * [EventSink] that allows putting events into the returned stream.
+   * This `EventSink` is only valid during the call to `onTimeout`.
+   *
+   * If the `onTimeout` function accepts two arguments, it is called with both

[floitsch, 2013/11/29 13:43:48]

As discussed. Let's not do this.

[Lasse Reichstein Nielsen, 2013/11/29 13:58:03]

Done.

+   * the `EventSink` and a function that allows canceling the input stream
+   * subscription. This will stop any further events from reaching the output
+   * stream.
+   *
+   * If `onTimeout` is omitted, a timeout will just put a [TimeoutException]
+   * into the error channel of the returned stream.
+   */
+  Stream timeout(Duration timeLimit, [Function onTimeout]) {

[floitsch, 2013/11/29 13:43:48]

make it named.

[Lasse Reichstein Nielsen, 2013/11/29 13:58:03]

Done.

+    StreamSubscription<T> subscription;
+    _StreamController controller;
+    Timer timer;
+    Zone outerZone = Zone.current;
+    Function timeout;
+    if (onTimeout == null) {
+      timeout = () {
+        controller.addError(new TimeoutException(/*"No stream event",*/
+                                                 timeLimit));
+      };
+    } else {
+      Zone zone = outerZone.fork();

[floitsch, 2013/11/29 13:43:48]

no need to fork.

[Lasse Reichstein Nielsen, 2013/11/29 13:58:03]

Done, by not doing!

+      if (onTimeout is ZoneBinaryCallback) {
+        onTimeout = zone.registerBinaryCallback(onTimeout);
+        _ControllerEventSinkWrapper wrapper =
+            new _ControllerEventSinkWrapper(null);
+        timeout = () {
+          wrapper._sink = controller;  // Only valid during call.
+          zone.runBinaryGuarded(onTimeout, wrapper, subscription.cancel);
+          wrapper._sink = null;
+        };
+      } else if (onTimeout is ZoneUnaryCallback) {
+        onTimeout = zone.registerUnaryCallback(onTimeout);
+        _ControllerEventSinkWrapper wrapper =
+            new _ControllerEventSinkWrapper(null);
+        timeout = () {
+          wrapper._sink = controller;  // Only valid during call.
+          zone.runUnaryGuarded(onTimeout, wrapper);
+          wrapper._sink = null;
+        };
+      } else {
+        onTimeout = zone.registerCallback(onTimeout);

[floitsch, 2013/11/29 13:43:48]

I would put that up to the onTimeout == null.

[Lasse Reichstein Nielsen, 2013/11/29 13:58:03]

What?
If onTimeout is null, why call registerCallback with it?

+        timeout = () { zone.runGuarded(onTimeout); };
+      }
+    }
+
+    void onData(T event) {
+      timer.cancel();

[floitsch, 2013/11/29 13:43:48]

This will be expensive.
For a first implementation ok, but we probably just want to reset an internal
number, and then look at that number when the actual timer triggers.
This way most onData won't cost anything.

[Lasse Reichstein Nielsen, 2013/11/29 13:58:03]

How will that work?
Have the initial timer run to its completion, and not start a new timer until
then, with the remaining time?

I can try doing that, but I'd prefer to put that in a different CL.

[floitsch, 2013/11/29 15:03:49]

yes. something like that.
me too.

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

[floitsch, 2013/11/29 13:43:48]

Don't we already have something like this?

[Lasse Reichstein Nielsen, 2013/11/29 13:58:03]

Not exactly, sadly. We have something that takes an _EventSink adapts it to an
EventSink, but not something that just exposes the EventSink nature of something
extending EventSink.

+ */
+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(); }
+}