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

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