Add Stream.timeout method.

sdk/lib/async/stream.dart

 // 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 dart.async;
 
 // -------------------------------------------------------------------
 // Core Stream types
 // -------------------------------------------------------------------
 
@@ skipping 631 matching lines @@
     return new _TakeStream(this, count);
   }
 
   /**
    * Forwards data events while [test] is successful.
    *
    * The returned stream provides the same events as this stream as long
    * 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
    * means that single-subscription (non-broadcast) streams are closed and
    * cannot be reused after a call to this method.
    */
   Stream<T> takeWhile(bool test(T element)) {
     return new _TakeWhileStream(this, test);
   }
 
@@ skipping 290 matching lines @@
         }
         index -= 1;
       },
       onError: future._completeError,
       onDone: () {
         future._completeError(new RangeError.value(index));
       },
       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;
+  }
 }
 
 /**
  * A control object for the subscription on a [Stream].
  *
  * When you subscribe on a [Stream] using [Stream.listen],
  * a [StreamSubscription] object is returned. This object
  * is used to later unsubscribe again, or to temporarily pause
  * the stream's events.
  */
@@ skipping 302 matching lines @@
    *
    * If you need to stop listening for values before the stream iterator is
    * automatically closed, you must call [cancel] to ensure that the stream
    * is properly closed.
    *
    * Returns a future if the cancel-operation is not completed synchronously.
    * Otherwise returns `null`.
    */
   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(); }
+}