Add Stream.timeout method.

Add Stream.timeout method.

R=floitsch@google.com

Committed: https://code.google.com/p/dart/source/detail?r=30895

[floitsch, 2013-11-29 13:43:47]

https://codereview.chromium.org/96473003/diff/40001/sdk/lib/async/stream.dart
File sdk/lib/async/stream.dart (right):

https://codereview.chromium.org/96473003/diff/40001/sdk/lib/async/stream.dart...
sdk/lib/async/stream.dart:978: * If the `onTimeout` function accepts two
arguments, it is called with both
As discussed. Let's not do this.

https://codereview.chromium.org/96473003/diff/40001/sdk/lib/async/stream.dart...
sdk/lib/async/stream.dart:986: Stream timeout(Duration timeLimit, [Function
onTimeout]) {
make it named.

https://codereview.chromium.org/96473003/diff/40001/sdk/lib/async/stream.dart...
sdk/lib/async/stream.dart:998: Zone zone = outerZone.fork();
no need to fork.

https://codereview.chromium.org/96473003/diff/40001/sdk/lib/async/stream.dart...
sdk/lib/async/stream.dart:1018: onTimeout = zone.registerCallback(onTimeout);
I would put that up to the onTimeout == null.

https://codereview.chromium.org/96473003/diff/40001/sdk/lib/async/stream.dart...
sdk/lib/async/stream.dart:1024: timer.cancel();
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.

https://codereview.chromium.org/96473003/diff/40001/sdk/lib/async/stream.dart...
sdk/lib/async/stream.dart:1383: * Wraps an [_EventSink] so it exposes only the
[EventSink] interface.
Don't we already have something like this?

https://codereview.chromium.org/96473003/diff/40001/tests/lib/async/stream_ti...
File tests/lib/async/stream_timeout_test.dart (right):

https://codereview.chromium.org/96473003/diff/40001/tests/lib/async/stream_ti...
tests/lib/async/stream_timeout_test.dart:118: new Timer.periodic(ms5, (timer) {
This could be flaky.
We have to guard against it.

https://codereview.chromium.org/96473003/diff/40001/tests/lib/async/stream_ti...
tests/lib/async/stream_timeout_test.dart:138: new Timer.periodic(ms5, (timer) {
ditto.

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

https://codereview.chromium.org/96473003/diff/40001/sdk/lib/async/stream.dart
File sdk/lib/async/stream.dart (right):

https://codereview.chromium.org/96473003/diff/40001/sdk/lib/async/stream.dart...
sdk/lib/async/stream.dart:978: * If the `onTimeout` function accepts two
arguments, it is called with both
On 2013/11/29 13:43:48, floitsch wrote:
> As discussed. Let's not do this.

Done.

https://codereview.chromium.org/96473003/diff/40001/sdk/lib/async/stream.dart...
sdk/lib/async/stream.dart:986: Stream timeout(Duration timeLimit, [Function
onTimeout]) {
On 2013/11/29 13:43:48, floitsch wrote:
> make it named.

Done.

https://codereview.chromium.org/96473003/diff/40001/sdk/lib/async/stream.dart...
sdk/lib/async/stream.dart:998: Zone zone = outerZone.fork();
Done, by not doing!

https://codereview.chromium.org/96473003/diff/40001/sdk/lib/async/stream.dart...
sdk/lib/async/stream.dart:1018: onTimeout = zone.registerCallback(onTimeout);
What?
If onTimeout is null, why call registerCallback with it?

https://codereview.chromium.org/96473003/diff/40001/sdk/lib/async/stream.dart...
sdk/lib/async/stream.dart:1024: timer.cancel();
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.

https://codereview.chromium.org/96473003/diff/40001/sdk/lib/async/stream.dart...
sdk/lib/async/stream.dart:1383: * Wraps an [_EventSink] so it exposes only the
[EventSink] interface.
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.

https://codereview.chromium.org/96473003/diff/40001/tests/lib/async/stream_ti...
File tests/lib/async/stream_timeout_test.dart (right):

https://codereview.chromium.org/96473003/diff/40001/tests/lib/async/stream_ti...
tests/lib/async/stream_timeout_test.dart:118: new Timer.periodic(ms5, (timer) {
It's using timers for both things. Will that not guarantee that
the 5ms timer always runs before the 500ms one?

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

https://codereview.chromium.org/96473003/diff/40001/sdk/lib/async/stream.dart
File sdk/lib/async/stream.dart (right):

https://codereview.chromium.org/96473003/diff/40001/sdk/lib/async/stream.dart...
sdk/lib/async/stream.dart:1024: timer.cancel();
On 2013/11/29 13:58:03, Lasse Reichstein Nielsen wrote:
> How will that work?
> Have the initial timer run to its completion, and not start a new timer until
> then, with the remaining time?
yes. something like that.
> 
> I can try doing that, but I'd prefer to put that in a different CL.
me too.

https://codereview.chromium.org/96473003/diff/40001/tests/lib/async/stream_ti...
File tests/lib/async/stream_timeout_test.dart (right):

https://codereview.chromium.org/96473003/diff/40001/tests/lib/async/stream_ti...
tests/lib/async/stream_timeout_test.dart:118: new Timer.periodic(ms5, (timer) {
On 2013/11/29 13:58:03, Lasse Reichstein Nielsen wrote:
> It's using timers for both things. Will that not guarantee that
> the 5ms timer always runs before the 500ms one?

Only, if you can guarantee that the periodic timer is restarted correctly wrt
the timeout timer.
Probably it is...

[Lasse Reichstein Nielsen, 2013-12-03 14:02:18]

PTAL

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

LGTM once the zones have been fixed.

https://chromiumcodereview.appspot.com/96473003/diff/110001/sdk/lib/async/str...
File sdk/lib/async/stream.dart (right):

https://chromiumcodereview.appspot.com/96473003/diff/110001/sdk/lib/async/str...
sdk/lib/async/stream.dart:1005: zone = Zone.current;
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:43]

https://chromiumcodereview.appspot.com/96473003/diff/110001/sdk/lib/async/str...
File sdk/lib/async/stream.dart (right):

https://chromiumcodereview.appspot.com/96473003/diff/110001/sdk/lib/async/str...
sdk/lib/async/stream.dart:1005: zone = Zone.current;
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;")

[Lasse Reichstein Nielsen, 2013-12-05 13:32:41]

Committed patchset #8 manually as r30895 (presubmit successful).

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

Noticed a small issue.

https://chromiumcodereview.appspot.com/96473003/diff/110001/sdk/lib/async/str...
File sdk/lib/async/stream.dart (right):

https://chromiumcodereview.appspot.com/96473003/diff/110001/sdk/lib/async/str...
sdk/lib/async/stream.dart:1005: zone = Zone.current;
On 2013/12/05 13:28:44, Lasse Reichstein Nielsen wrote:
> 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;") 

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

https://chromiumcodereview.appspot.com/96473003/diff/110001/sdk/lib/async/str...
sdk/lib/async/stream.dart:1017: zone.runUnaryGuarded(onTimeout, wrapper);
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]

https://chromiumcodereview.appspot.com/96473003/diff/110001/sdk/lib/async/str...
File sdk/lib/async/stream.dart (right):

https://chromiumcodereview.appspot.com/96473003/diff/110001/sdk/lib/async/str...
sdk/lib/async/stream.dart:1017: zone.runUnaryGuarded(onTimeout, wrapper);
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.