Make StreamSubscription be the active part of a stream.

Make StreamSubscription be the active part of a stream.

It is the one keeping the buffer of events, not the stream.

R=ajohnsen@google.com, floitsch@google.com, jmesserly@google.com

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

[Lasse Reichstein Nielsen, 2013-05-08 12:29:30]

FYI, not done yet.

[Lasse Reichstein Nielsen, 2013-05-22 12:53:38]

This is getting close to working.
Notice that you can no longer listen on a stream more than once. Broadcast
streams start listening on the first listen, and only allow one batch of
listeners. They die as soon as there are no listeners again.

Bob: Please take a look at the changes to pub. There is still one new error in
error_group_test/"with a single stream should pass a signaled exception to the
stream if it has a listener and should unsubscribe that stream". With the change
to ErrorGroup to not use throwDelayed, it now breaks the entire test.

Søren/Anders, please take a look at the the io-lib changes.

[Lasse Reichstein Nielsen, 2013-05-22 12:56:08]

Please take looks.

[Anders Johnsen, 2013-05-22 13:07:08]

IO stuff LGTM

https://codereview.chromium.org/14753009/diff/6001/runtime/bin/socket_patch.dart
File runtime/bin/socket_patch.dart (right):

https://codereview.chromium.org/14753009/diff/6001/runtime/bin/socket_patch.d...
runtime/bin/socket_patch.dart:812: assert(!!!paused);
!!!?

https://codereview.chromium.org/14753009/diff/6001/runtime/bin/socket_patch.d...
runtime/bin/socket_patch.dart:850: if (subscription == null) return;
This line (and the comment) can be removed now, due to async future completion.

https://codereview.chromium.org/14753009/diff/6001/runtime/bin/socket_patch.d...
runtime/bin/socket_patch.dart:851: paused = true;
Can you try without these pause?

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/io/http_impl.dart
File sdk/lib/io/http_impl.dart (right):

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/io/http_impl.dart#...
sdk/lib/io/http_impl.dart:595: Future result = _completer.future;
Nooo... :)

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/io/http_parser.dart
File sdk/lib/io/http_parser.dart (left):

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/io/http_parser.dar...
sdk/lib/io/http_parser.dart:931: void update(bool pause) {
Ty!

https://codereview.chromium.org/14753009/diff/6001/tests/standalone/io/http_b...
File tests/standalone/io/http_body_test.dart (right):

https://codereview.chromium.org/14753009/diff/6001/tests/standalone/io/http_b...
tests/standalone/io/http_body_test.dart:171: //  test("text/plain",
"body".codeUnits, "body", "text");
No :)

https://codereview.chromium.org/14753009/diff/6001/tests/standalone/io/http_b...
tests/standalone/io/http_body_test.dart:204: return;
No :)

https://codereview.chromium.org/14753009/diff/6001/tests/standalone/io/http_b...
tests/standalone/io/http_body_test.dart:324: //  testHttpClientResponseBody();
No :)

https://codereview.chromium.org/14753009/diff/6001/tests/standalone/io/http_m...
File tests/standalone/io/http_multipart_test.dart (right):

https://codereview.chromium.org/14753009/diff/6001/tests/standalone/io/http_m...
tests/standalone/io/http_multipart_test.dart:119: return;
No :)

[Bob Nystrom, 2013-05-22 16:23:46]

Added Nathan to review the scheduled_test stuff too.

https://codereview.chromium.org/14753009/diff/6001/pkg/scheduled_test/lib/sch...
File pkg/scheduled_test/lib/scheduled_process.dart (right):

https://codereview.chromium.org/14753009/diff/6001/pkg/scheduled_test/lib/sch...
pkg/scheduled_test/lib/scheduled_process.dart:250: }
Instead of doing this by hand, use new Trace.current() in the stack_trace
package.

https://codereview.chromium.org/14753009/diff/6001/pkg/scheduled_test/lib/sch...
pkg/scheduled_test/lib/scheduled_process.dart:273: return
streamIterator.moveNext().then((bool hasNext) {
Don't bother type annotating parameters in fn expressions.

https://codereview.chromium.org/14753009/diff/6001/pkg/scheduled_test/lib/sch...
pkg/scheduled_test/lib/scheduled_process.dart:276: return
collectAll(accumulator);
If the stream has a lot of output, this can build a very long chain of futures.
They're async, so they won't blow the stack, but I've seen long future chains
exhaust the heap or consume a lot of CPU traversing the chain.

Maybe we should look to add something reduce-like to dart:async specifically to
address this? You can handle it yourself using a completer, but it's tedious and
error-prone to do so.

https://codereview.chromium.org/14753009/diff/6001/pkg/scheduled_test/lib/sch...
pkg/scheduled_test/lib/scheduled_process.dart:278: return
accumulator.join("\n");
A StringBuffer seems like a more natural fit for this than a List.

https://codereview.chromium.org/14753009/diff/6001/pkg/scheduled_test/lib/sch...
pkg/scheduled_test/lib/scheduled_process.dart:292: List<String> accumulator =
<String>[];
Unused.

[floitsch, 2013-05-22 16:26:29]

Still need to look at the test changes, but so far LGTM.

https://codereview.chromium.org/14753009/diff/6001/pkg/scheduled_test/lib/sch...
File pkg/scheduled_test/lib/scheduled_process.dart (right):

https://codereview.chromium.org/14753009/diff/6001/pkg/scheduled_test/lib/sch...
pkg/scheduled_test/lib/scheduled_process.dart:243: // And it works at all, which
streamFirst doesn't.
Reword. This could be read like an insult.

https://codereview.chromium.org/14753009/diff/6001/pkg/scheduled_test/lib/sch...
pkg/scheduled_test/lib/scheduled_process.dart:292: List<String> accumulator =
<String>[];
Looks unused.

https://codereview.chromium.org/14753009/diff/6001/pkg/scheduled_test/lib/src...
File pkg/scheduled_test/lib/src/mock_clock.dart (right):

https://codereview.chromium.org/14753009/diff/6001/pkg/scheduled_test/lib/src...
pkg/scheduled_test/lib/src/mock_clock.dart:57: /// Creates a new stream
controller and its stream as broadcast stream.
The behavior is not the same anymore.
When there are pending events the new version discards pending events.
Make sure the new behavior is acceptable.

https://codereview.chromium.org/14753009/diff/6001/pkg/scheduled_test/lib/src...
pkg/scheduled_test/lib/src/mock_clock.dart:83: if (_onTickController == null)
return;
the _onTickController can never be null. I think the old code should be fine.

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/_internal/pub/test...
File sdk/lib/_internal/pub/test/error_group_test.dart (left):

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/_internal/pub/test...
sdk/lib/_internal/pub/test/error_group_test.dart:258: 
keep line.

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/_internal/pub/test...
File sdk/lib/_internal/pub/test/error_group_test.dart (right):

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/_internal/pub/test...
sdk/lib/_internal/pub/test/error_group_test.dart:220: expect(iter.moveNext(),
completion(isFalse));
I think the `completion` is called too late. The test will not see that this is
an asynchronous test before it has already shut down.

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/_internal/pub/test...
File sdk/lib/_internal/pub/test/package_files_test.dart (right):

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/_internal/pub/test...
sdk/lib/_internal/pub/test/package_files_test.dart:1: // Copyright (c) 2012, the
Dart project authors.  Please see the AUTHORS file
Keep file as is. I don't see need for change.

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

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream.dart#...
sdk/lib/async/stream.dart:1176: class _EventSinkAdapter<T> extends EventSink<T>
{
I would prefer "implements" here. We really use EventSink like an interface.

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream_contr...
File sdk/lib/async/stream_controller.dart (left):

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream_contr...
sdk/lib/async/stream_controller.dart:57: onListen, onPause, onResume, onCancel);
You will have to merge this. I removed lines 53 to 57.

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream_contr...
File sdk/lib/async/stream_controller.dart (right):

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream_contr...
sdk/lib/async/stream_controller.dart:62: // Whether we have sent a "done" event.
"Wether" is not right.
Also, what does "0" mean?

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream_contr...
sdk/lib/async/stream_controller.dart:122: /** Whether there are currently any
subscribers on the [Stream]. */
there is currently a subscriber on the [Stream].

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream_contr...
sdk/lib/async/stream_controller.dart:140: * If the subscription has requested to
be unsubscribed on errors,
Should we keep this here? It's not really a Controller thing.

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream_contr...
sdk/lib/async/stream_controller.dart:160: * The "done" message should be sent at
most once by a stream, and it
Too many "should"s.
Maybe:
Closes this controller.

It is valid to close a controller multiple times. Only the first invocation has
an effect.

Upon the first closing the controller will send (or enqueue) a done event.

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream_contr...
sdk/lib/async/stream_controller.dart:185: _pendingEvents = null;  // These have
been taken over by the stream.
by the subscription.

I would prefer if we transferred the pendingEvents explicitly here. It's almost
impossible to see at the construction site.
Something like: 
subscription._transferPendingEvents(_pendingEvents);
_pendingEvents = null;

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream_impl....
File sdk/lib/async/stream_impl.dart (right):

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream_impl....
sdk/lib/async/stream_impl.dart:59: * The user should not add new eventswhen the
subscription requests a paused,
space missing.

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream_impl....
sdk/lib/async/stream_impl.dart:86: static const int _STATE_CANCELLED = 8;
_STATE_CANCELED

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream_impl....
sdk/lib/async/stream_impl.dart:176: print("BAD THROW: \n$s");
So _onCancel is not allowed to throw? Should this be an uncaught error?

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream_impl....
sdk/lib/async/stream_impl.dart:199: bool get _cancelled => (_state &
_STATE_CANCELLED) != 0;
isCanceled

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream_impl....
sdk/lib/async/stream_impl.dart:205: !_isPaused && (_pending == null ||
_pending.isEmpty);
shouldn't _hasPending do the null-check too?

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream_impl....
sdk/lib/async/stream_impl.dart:316: try {
move the `try`s into a separate function?

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream_impl....
sdk/lib/async/stream_impl.dart:385: * After cancelling, no further callbacks
will happen.
separate with more new lines.

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream_impl....
sdk/lib/async/stream_impl.dart:397: while (true) {
Add comment why we need a loop.

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream_impl....
sdk/lib/async/stream_impl.dart:523: _iterationState = _STATE_CANCELLED;  // Will
send a single done after this.
long line.

I don't think it makes sense to send a done when there was an error.

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream_impl....
sdk/lib/async/stream_impl.dart:693: * Timer set when pending events are
scheduled for execution.
Update comments.

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream_impl....
sdk/lib/async/stream_impl.dart:794: class _MultiplexerSubscription<T> extends
_BufferingStreamSubscription<T>
Add comment what this class does.

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream_impl....
sdk/lib/async/stream_impl.dart:872: class _SingleStreamMultiplexer<T> extends
Stream<T>
Add comment.

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream_impl....
sdk/lib/async/stream_impl.dart:981: void _add(T data) {
Why is this not handled by the bufferingStreamSubscription superclass?
It checks already if _canFire.

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream_impl....
sdk/lib/async/stream_impl.dart:992: if (_isFiring || _hasPending) {
ditto.

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream_impl....
sdk/lib/async/stream_impl.dart:1002: if (_isFiring || _hasPending) {
ditto.

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream_impl....
sdk/lib/async/stream_impl.dart:1042: // TODO(lrn): Keep a one (or two) element
buffer to avoid pausing when not
I guess the ideal solution would be if the constructor allowed to give the
buffer-size.

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream_impl....
sdk/lib/async/stream_impl.dart:1096: // We have cancelOnError: true, so the
subscription is cancelled.
canceled.

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream_pipe....
File sdk/lib/async/stream_pipe.dart (right):

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream_pipe....
sdk/lib/async/stream_pipe.dart:109: // is cancelled or closed after the first,
so we just ignore remaining events.
canceled.

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/io/http_impl.dart
File sdk/lib/io/http_impl.dart (right):

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/io/http_impl.dart#...
sdk/lib/io/http_impl.dart:595: Future result = _completer.future;
Apparently this was necessary.
Add comment. (it's not obvious)

https://codereview.chromium.org/14753009/diff/6001/tests/corelib/list_test.dart
File tests/corelib/list_test.dart (right):

https://codereview.chromium.org/14753009/diff/6001/tests/corelib/list_test.da...
tests/corelib/list_test.dart:17: // toList returns a List<int>.
unrelated change?

[nweiz, 2013-05-22 18:42:41]

https://codereview.chromium.org/14753009/diff/6001/pkg/scheduled_test/lib/sch...
File pkg/scheduled_test/lib/scheduled_process.dart (right):

https://codereview.chromium.org/14753009/diff/6001/pkg/scheduled_test/lib/sch...
pkg/scheduled_test/lib/scheduled_process.dart:42: StreamIterator<String>
_stdout;
I'm not sure I understand why using a StreamIterator here is better than using a
Stream. It makes the implementation of [remainingStdout] and [remainingStderr]
more complicated, and I don't see a corresponding benefit.

https://codereview.chromium.org/14753009/diff/6001/pkg/scheduled_test/lib/sch...
pkg/scheduled_test/lib/scheduled_process.dart:244: Future<String>
_streamIteratorFirst(StreamIterator<String> streamIterator) {
This belongs in utils.dart.

https://codereview.chromium.org/14753009/diff/6001/pkg/scheduled_test/lib/sch...
pkg/scheduled_test/lib/scheduled_process.dart:271: static Future<String>
_concatRest(StreamIterator<String> streamIterator) {
This also belongs in utils.dart.

https://codereview.chromium.org/14753009/diff/6001/pkg/scheduled_test/lib/src...
File pkg/scheduled_test/lib/src/mock_clock.dart (right):

https://codereview.chromium.org/14753009/diff/6001/pkg/scheduled_test/lib/src...
pkg/scheduled_test/lib/src/mock_clock.dart:64: }
I don't understand these changes. Why are we creating a new stream when a user
cancels their subscription? What happens to other users with subscriptions to
the old stream?

[floitsch, 2013-05-22 20:23:35]

https://codereview.chromium.org/14753009/diff/6001/pkg/scheduled_test/lib/sch...
File pkg/scheduled_test/lib/scheduled_process.dart (right):

https://codereview.chromium.org/14753009/diff/6001/pkg/scheduled_test/lib/sch...
pkg/scheduled_test/lib/scheduled_process.dart:42: StreamIterator<String>
_stdout;
On 2013/05/22 18:42:41, nweiz wrote:
> I'm not sure I understand why using a StreamIterator here is better than using
a
> Stream. It makes the implementation of [remainingStdout] and [remainingStderr]
> more complicated, and I don't see a corresponding benefit.

The moment you cancel a stream it is dead. We have had a bug for some time where
you could still subscribe to it.
This CL fixes the bug. Unfortunately there is code that relied on the broken
behavior. The StreamIterator class provides an interface that usually allows to
simulate the old behavior.

[floitsch, 2013-05-23 18:38:22]

https://codereview.chromium.org/14753009/diff/6001/tests/lib/async/stream_sta...
File tests/lib/async/stream_state_nonzero_timer_test.dart (right):

https://codereview.chromium.org/14753009/diff/6001/tests/lib/async/stream_sta...
tests/lib/async/stream_state_nonzero_timer_test.dart:84: return;
spurious 'return'.

https://codereview.chromium.org/14753009/diff/23001/pkg/scheduled_test/lib/sc...
File pkg/scheduled_test/lib/scheduled_process.dart (right):

https://codereview.chromium.org/14753009/diff/23001/pkg/scheduled_test/lib/sc...
pkg/scheduled_test/lib/scheduled_process.dart:243: * Returns the first element
of a [StreamIterator].
Remove and use streamIteratorFirst from utils ?

https://codereview.chromium.org/14753009/diff/23001/pkg/scheduled_test/lib/sc...
pkg/scheduled_test/lib/scheduled_process.dart:269: static Future<String>
_concatRest(StreamIterator<String> streamIterator) {
ditto.

https://codereview.chromium.org/14753009/diff/23001/pkg/scheduled_test/lib/sc...
pkg/scheduled_test/lib/scheduled_process.dart:290: List<String> accumulator =
<String>[];
unused.

https://codereview.chromium.org/14753009/diff/23001/pkg/scheduled_test/lib/sr...
File pkg/scheduled_test/lib/src/mock_clock.dart (right):

https://codereview.chromium.org/14753009/diff/23001/pkg/scheduled_test/lib/sr...
pkg/scheduled_test/lib/src/mock_clock.dart:54: StreamController<int> controller;
We should really provide a StreamController that does this for you...

https://codereview.chromium.org/14753009/diff/23001/pkg/scheduled_test/lib/sr...
File pkg/scheduled_test/lib/src/utils.dart (right):

https://codereview.chromium.org/14753009/diff/23001/pkg/scheduled_test/lib/sr...
pkg/scheduled_test/lib/src/utils.dart:118:
streamIterator.moveNext().then((hasNext) {
We could consider having this on the iterator. It seems like a common operation.

[Lasse Reichstein Nielsen, 2013-05-24 06:02:48]

https://codereview.chromium.org/14753009/diff/6001/pkg/scheduled_test/lib/sch...
File pkg/scheduled_test/lib/scheduled_process.dart (right):

https://codereview.chromium.org/14753009/diff/6001/pkg/scheduled_test/lib/sch...
pkg/scheduled_test/lib/scheduled_process.dart:42: StreamIterator<String>
_stdout;
Exactly. 
The current code subscribed to the same stream more than once (using streamFirst
one or more time). That does not work any more.

https://codereview.chromium.org/14753009/diff/6001/pkg/scheduled_test/lib/sch...
pkg/scheduled_test/lib/scheduled_process.dart:243: // And it works at all, which
streamFirst doesn't.
Ack, working comment left in :)

https://codereview.chromium.org/14753009/diff/6001/pkg/scheduled_test/lib/sch...
pkg/scheduled_test/lib/scheduled_process.dart:244: Future<String>
_streamIteratorFirst(StreamIterator<String> streamIterator) {
Ok, will move it.

https://codereview.chromium.org/14753009/diff/6001/pkg/scheduled_test/lib/sch...
pkg/scheduled_test/lib/scheduled_process.dart:250: }
Will do.

https://codereview.chromium.org/14753009/diff/6001/pkg/scheduled_test/lib/sch...
pkg/scheduled_test/lib/scheduled_process.dart:271: static Future<String>
_concatRest(StreamIterator<String> streamIterator) {
On 2013/05/22 18:42:41, nweiz wrote:
> This also belongs in utils.dart.

Done.

https://codereview.chromium.org/14753009/diff/6001/pkg/scheduled_test/lib/sch...
pkg/scheduled_test/lib/scheduled_process.dart:273: return
streamIterator.moveNext().then((bool hasNext) {
ok.
Grudgingly.
How else would I know the type of hasNext from looking at the code? :(

https://codereview.chromium.org/14753009/diff/6001/pkg/scheduled_test/lib/sch...
pkg/scheduled_test/lib/scheduled_process.dart:276: return
collectAll(accumulator);
I've rewritten this to not chain the futures, and just complete a completer at
the end.

https://codereview.chromium.org/14753009/diff/6001/pkg/scheduled_test/lib/sch...
pkg/scheduled_test/lib/scheduled_process.dart:278: return
accumulator.join("\n");
Good point.

https://codereview.chromium.org/14753009/diff/6001/pkg/scheduled_test/lib/sch...
pkg/scheduled_test/lib/scheduled_process.dart:292: List<String> accumulator =
<String>[];
Well spotted.

https://codereview.chromium.org/14753009/diff/6001/pkg/scheduled_test/lib/src...
File pkg/scheduled_test/lib/src/mock_clock.dart (right):

https://codereview.chromium.org/14753009/diff/6001/pkg/scheduled_test/lib/src...
pkg/scheduled_test/lib/src/mock_clock.dart:64: }
The stream-controller only received the cancel when all subscribers of the
broadcast stream are gone.
At that point, the stream is dead, and any new subscribers will have to listen
on a new stream.

I think this should instead be done with one controller per listener. You don't
get events that fires before you listen, you get all events that fire while you
listen, and then your stream is dead when you stop listening.
I've implemented this. It is similar (at least in spirit) to how DOM streams are
implemented - each time you read the .onTick property, you get a new stream.

https://codereview.chromium.org/14753009/diff/6001/pkg/scheduled_test/lib/src...
pkg/scheduled_test/lib/src/mock_clock.dart:83: if (_onTickController == null)
return;
Good point.

https://codereview.chromium.org/14753009/diff/6001/runtime/bin/socket_patch.dart
File runtime/bin/socket_patch.dart (right):

https://codereview.chromium.org/14753009/diff/6001/runtime/bin/socket_patch.d...
runtime/bin/socket_patch.dart:812: assert(!!!paused);
!!!!!
Deliberately made stupid to mark it as debug code :)

https://codereview.chromium.org/14753009/diff/6001/runtime/bin/socket_patch.d...
runtime/bin/socket_patch.dart:850: if (subscription == null) return;
YEY

https://codereview.chromium.org/14753009/diff/6001/runtime/bin/socket_patch.d...
runtime/bin/socket_patch.dart:851: paused = true;
I'll try.

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/_internal/pub/test...
File sdk/lib/_internal/pub/test/error_group_test.dart (left):

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/_internal/pub/test...
sdk/lib/_internal/pub/test/error_group_test.dart:258: 
ok.

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/_internal/pub/test...
File sdk/lib/_internal/pub/test/error_group_test.dart (right):

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/_internal/pub/test...
sdk/lib/_internal/pub/test/error_group_test.dart:220: expect(iter.moveNext(),
completion(isFalse));
Then how do you write a test that expects an async result and can only continue
after that?

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/_internal/pub/test...
File sdk/lib/_internal/pub/test/package_files_test.dart (right):

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/_internal/pub/test...
sdk/lib/_internal/pub/test/package_files_test.dart:1: // Copyright (c) 2012, the
Dart project authors.  Please see the AUTHORS file
Will do. Leftovers from trying to find out why the test failed.

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

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream.dart#...
sdk/lib/async/stream.dart:1176: class _EventSinkAdapter<T> extends EventSink<T>
{
Done.

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream_contr...
File sdk/lib/async/stream_controller.dart (left):

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream_contr...
sdk/lib/async/stream_controller.dart:57: onListen, onPause, onResume, onCancel);
ACK.

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream_contr...
File sdk/lib/async/stream_controller.dart (right):

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream_contr...
sdk/lib/async/stream_controller.dart:62: // Whether we have sent a "done" event.
On 2013/05/22 16:26:29, floitsch wrote:
> "Wether" is not right.

I see "Whether"?

> Also, what does "0" mean?

Will name a constant _STATE_OPEN or _STATE_INITIAL.

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream_contr...
sdk/lib/async/stream_controller.dart:122: /** Whether there are currently any
subscribers on the [Stream]. */
Done.

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream_contr...
sdk/lib/async/stream_controller.dart:140: * If the subscription has requested to
be unsubscribed on errors,
I'm fine with removing it.

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream_contr...
sdk/lib/async/stream_controller.dart:160: * The "done" message should be sent at
most once by a stream, and it
sounds fine.

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream_contr...
sdk/lib/async/stream_controller.dart:185: _pendingEvents = null;  // These have
been taken over by the stream.
ok, will change to not pass them in the constructor.

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream_impl....
File sdk/lib/async/stream_impl.dart (right):

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream_impl....
sdk/lib/async/stream_impl.dart:59: * The user should not add new eventswhen the
subscription requests a paused,
On 2013/05/22 16:26:29, floitsch wrote:
> space missing.

Done.

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream_impl....
sdk/lib/async/stream_impl.dart:86: static const int _STATE_CANCELLED = 8;
Bowdlerizing constants! :(

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream_impl....
sdk/lib/async/stream_impl.dart:176: print("BAD THROW: \n$s");
_onCancel is an internal method only. It should never throw (documented above
_onPause, but I should probably put documentation on all methods). This
try/catch is just for debugging purposes, to make sure it doesn't actually throw
anyway. I'll remove it completely (ditto for other "BAD THROW" parts)

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream_impl....
sdk/lib/async/stream_impl.dart:199: bool get _cancelled => (_state &
_STATE_CANCELLED) != 0;
On 2013/05/22 16:26:29, floitsch wrote:
> isCanceled

Done.

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream_impl....
sdk/lib/async/stream_impl.dart:205: !_isPaused && (_pending == null ||
_pending.isEmpty);
_hasPending is accessing a cached bit in the state field. It is set when a
pending event is added, and removed *after* the last pending event has been
fired. Having it there allows checking for whether an event can be fired
directly by a single check on _state.

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream_impl....
sdk/lib/async/stream_impl.dart:316: try {
Rather not. It would only be used for _send{Data,Error,Done}, so an extra
function call with a closure body is overkill for avoiding four duplicate lines.

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream_impl....
sdk/lib/async/stream_impl.dart:385: * After cancelling, no further callbacks
will happen.
On 2013/05/22 16:26:29, floitsch wrote:
> separate with more new lines.

Done.

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream_impl....
sdk/lib/async/stream_impl.dart:397: while (true) {
Ok.

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream_impl....
sdk/lib/async/stream_impl.dart:523: _iterationState = _STATE_CANCELLED;  // Will
send a single done after this.
Good point. Some listeners might be left hanging, but there shouldn't be an
error here, so it's likely that the program is borked anyway.

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream_impl....
sdk/lib/async/stream_impl.dart:693: * Timer set when pending events are
scheduled for execution.
On 2013/05/22 16:26:29, floitsch wrote:
> Update comments.

Done.

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream_impl....
sdk/lib/async/stream_impl.dart:794: class _MultiplexerSubscription<T> extends
_BufferingStreamSubscription<T>
On 2013/05/22 16:26:29, floitsch wrote:
> Add comment what this class does.

Done.

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream_impl....
sdk/lib/async/stream_impl.dart:872: class _SingleStreamMultiplexer<T> extends
Stream<T>
On 2013/05/22 16:26:29, floitsch wrote:
> Add comment.

Done.

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream_impl....
sdk/lib/async/stream_impl.dart:981: void _add(T data) {
This isn't a StreamSubscription - it's a Stream, maintaining its multiple
simultaneous subscribers.

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream_impl....
sdk/lib/async/stream_impl.dart:992: if (_isFiring || _hasPending) {
ditto too.

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream_impl....
sdk/lib/async/stream_impl.dart:1042: // TODO(lrn): Keep a one (or two) element
buffer to avoid pausing when not
I don't want this to be about buffering, it should just be a single, simple
StreamIterator. I just worry that pausing after each event is too heavy-handed,
and giving it a one-element buffer would allow it to avoid that in the cases
where elements are used immediately. Only if a new element comes in before the
previous one has been processed, and moveNext called, do we need to pause and
store the extra element - so a one element cache is enough.

It's optimization only.

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream_impl....
sdk/lib/async/stream_impl.dart:1096: // We have cancelOnError: true, so the
subscription is cancelled.
On 2013/05/22 16:26:29, floitsch wrote:
> canceled.

Done.

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream_pipe....
File sdk/lib/async/stream_pipe.dart (right):

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream_pipe....
sdk/lib/async/stream_pipe.dart:109: // is cancelled or closed after the first,
so we just ignore remaining events.
On 2013/05/22 16:26:29, floitsch wrote:
> canceled.

Done.

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/io/http_impl.dart
File sdk/lib/io/http_impl.dart (right):

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/io/http_impl.dart#...
sdk/lib/io/http_impl.dart:595: Future result = _completer.future;
It should not be necessary any more after I prevented sending events during the
onListen callback (stupid bug). Will change it back.

https://codereview.chromium.org/14753009/diff/6001/tests/corelib/list_test.dart
File tests/corelib/list_test.dart (right):

https://codereview.chromium.org/14753009/diff/6001/tests/corelib/list_test.da...
tests/corelib/list_test.dart:17: // toList returns a List<int>.
Completely.

https://codereview.chromium.org/14753009/diff/6001/tests/standalone/io/http_b...
File tests/standalone/io/http_body_test.dart (right):

https://codereview.chromium.org/14753009/diff/6001/tests/standalone/io/http_b...
tests/standalone/io/http_body_test.dart:171: //  test("text/plain",
"body".codeUnits, "body", "text");
Aaah, please?
The commented out tests were the ones that didn't fail during debugging, so
maybe I should reenable them :)

https://codereview.chromium.org/14753009/diff/6001/tests/standalone/io/http_b...
tests/standalone/io/http_body_test.dart:204: return;
On 2013/05/22 13:07:08, Anders Johnsen wrote:
> No :)

Done.

https://codereview.chromium.org/14753009/diff/6001/tests/standalone/io/http_m...
File tests/standalone/io/http_multipart_test.dart (right):

https://codereview.chromium.org/14753009/diff/6001/tests/standalone/io/http_m...
tests/standalone/io/http_multipart_test.dart:119: return;
On 2013/05/22 13:07:08, Anders Johnsen wrote:
> No :)

Done.

[Lasse Reichstein Nielsen, 2013-05-24 11:13:29]

All tests run.
Working in dart2js too.
Had to change ObserveMixin to avoid listening twice to the same stream.

https://codereview.chromium.org/14753009/diff/23001/pkg/scheduled_test/lib/sr...
File pkg/scheduled_test/lib/src/utils.dart (right):

https://codereview.chromium.org/14753009/diff/23001/pkg/scheduled_test/lib/sr...
pkg/scheduled_test/lib/src/utils.dart:118:
streamIterator.moveNext().then((hasNext) {
What would the operation be?
 Future StreamIterator.moveNext(action(bool hasNext)); ?

[floitsch, 2013-05-24 13:53:41]

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/_internal/pub/test...
File sdk/lib/_internal/pub/test/error_group_test.dart (right):

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/_internal/pub/test...
sdk/lib/_internal/pub/test/error_group_test.dart:220: expect(iter.moveNext(),
completion(isFalse));
On 2013/05/24 06:02:49, Lasse Reichstein Nielsen wrote:
> Then how do you write a test that expects an async result and can only
continue
> after that?

expectAsync the first asynchronous part. then the next ...

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream_contr...
File sdk/lib/async/stream_controller.dart (right):

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream_contr...
sdk/lib/async/stream_controller.dart:62: // Whether we have sent a "done" event.
On 2013/05/24 06:02:49, Lasse Reichstein Nielsen wrote:
> On 2013/05/22 16:26:29, floitsch wrote:
> > "Wether" is not right.
> 
> I see "Whether"?
I meant that this doesn't look a boolean. "Whether" is used for booleans.
> 
> > Also, what does "0" mean?
> 
> Will name a constant _STATE_OPEN or _STATE_INITIAL.
>

https://codereview.chromium.org/14753009/diff/23001/pkg/scheduled_test/lib/sr...
File pkg/scheduled_test/lib/src/utils.dart (right):

https://codereview.chromium.org/14753009/diff/23001/pkg/scheduled_test/lib/sr...
pkg/scheduled_test/lib/src/utils.dart:118:
streamIterator.moveNext().then((hasNext) {
On 2013/05/24 11:13:30, Lasse Reichstein Nielsen wrote:
> What would the operation be?
>  Future StreamIterator.moveNext(action(bool hasNext)); ?

I meant the "collectAll".

[floitsch, 2013-05-24 15:53:17]

LGTM.

https://codereview.chromium.org/14753009/diff/44002/pkg/scheduled_test/lib/sr...
File pkg/scheduled_test/lib/src/mock_clock.dart (right):

https://codereview.chromium.org/14753009/diff/44002/pkg/scheduled_test/lib/sr...
pkg/scheduled_test/lib/src/mock_clock.dart:51: Set<StreamController>
_subscriptions = new Set<StreamController>();
Add TODO that we want to replace this with something more efficient.

https://codereview.chromium.org/14753009/diff/44002/sdk/lib/_internal/pub/tes...
File sdk/lib/_internal/pub/test/error_group_test.dart (right):

https://codereview.chromium.org/14753009/diff/44002/sdk/lib/_internal/pub/tes...
sdk/lib/_internal/pub/test/error_group_test.dart:214:
iter.moveNext().then((hasNext) {
iter.moveNext().then(expectAsync1((hasNext) {
  ...

https://codereview.chromium.org/14753009/diff/44002/sdk/lib/_internal/pub/tes...
sdk/lib/_internal/pub/test/error_group_test.dart:248: expect(new Future(() {
Does this part of the test still make sense now?

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

https://codereview.chromium.org/14753009/diff/44002/sdk/lib/async/stream.dart...
sdk/lib/async/stream.dart:1215: * Only valid when the future returned by
[moveNext] completes with `true`
"valid" or non-null?

https://codereview.chromium.org/14753009/diff/44002/sdk/lib/async/stream_cont...
File sdk/lib/async/stream_controller.dart (right):

https://codereview.chromium.org/14753009/diff/44002/sdk/lib/async/stream_cont...
sdk/lib/async/stream_controller.dart:183: void
_recordListen(_BufferingStreamSubscription subscription) {
add function comment.

https://codereview.chromium.org/14753009/diff/44002/sdk/lib/async/stream_cont...
sdk/lib/async/stream_controller.dart:188: subscription._guardCallback(() {
Explain why you don't call _runGuarded directly.

https://codereview.chromium.org/14753009/diff/44002/sdk/lib/async/stream_pipe...
File sdk/lib/async/stream_pipe.dart (right):

https://codereview.chromium.org/14753009/diff/44002/sdk/lib/async/stream_pipe...
sdk/lib/async/stream_pipe.dart:273: class _TakeStream<T> extends
_ForwardingStream<T, T> {
In theory it should be the subscription that does the counting. Since the stream
is single subscription it doesn't matter, though.

https://codereview.chromium.org/14753009/diff/44002/sdk/lib/io/mime_multipart...
File sdk/lib/io/mime_multipart_parser.dart (right):

https://codereview.chromium.org/14753009/diff/44002/sdk/lib/io/mime_multipart...
sdk/lib/io/mime_multipart_parser.dart:106: onCancel: () {
add comment, that we need a closure, because _subscription is not set yet.

[Bob Nystrom, 2013-05-24 15:54:37]

https://codereview.chromium.org/14753009/diff/6001/pkg/scheduled_test/lib/sch...
File pkg/scheduled_test/lib/scheduled_process.dart (right):

https://codereview.chromium.org/14753009/diff/6001/pkg/scheduled_test/lib/sch...
pkg/scheduled_test/lib/scheduled_process.dart:273: return
streamIterator.moveNext().then((bool hasNext) {
On 2013/05/24 06:02:49, Lasse Reichstein Nielsen wrote:
> ok.
> Grudgingly.
> How else would I know the type of hasNext from looking at the code? :(

1. You know what you're passing that callback to, so you probably know the fn
type it expects.
2. It's name starts with "has" -> probably a bool.
3. The type doesn't matter that much, its name tells you what it is.

[Jennifer Messerly, 2013-05-24 18:47:55]

MDV changes lgtm

https://codereview.chromium.org/14753009/diff/44002/sdk/lib/mdv_observe_impl/...
File sdk/lib/mdv_observe_impl/mdv_observe_impl.dart (right):

https://codereview.chromium.org/14753009/diff/44002/sdk/lib/mdv_observe_impl/...
sdk/lib/mdv_observe_impl/mdv_observe_impl.dart:97: // TODO(jmesserly): make
"changes" immutable
keep this comment by the "observer.add(changes);" line?

[nweiz, 2013-05-24 20:12:30]

https://codereview.chromium.org/14753009/diff/6001/pkg/scheduled_test/lib/sch...
File pkg/scheduled_test/lib/scheduled_process.dart (right):

https://codereview.chromium.org/14753009/diff/6001/pkg/scheduled_test/lib/sch...
pkg/scheduled_test/lib/scheduled_process.dart:42: StreamIterator<String>
_stdout;
On 2013/05/24 06:02:49, Lasse Reichstein Nielsen wrote:
> Exactly. 
> The current code subscribed to the same stream more than once (using
streamFirst
> one or more time). That does not work any more.

Okay, I suppose that makes sense. It seems a little weird to me that having an
Iterator-style API and supporting multiple subscriptions are conflated concepts,
but that's a different discussion and I don't feel too strongly about it.

If streamFirst is completely broken now, please remove it from utils.dart.

https://codereview.chromium.org/14753009/diff/6001/pkg/scheduled_test/lib/sch...
pkg/scheduled_test/lib/scheduled_process.dart:276: return
collectAll(accumulator);
On 2013/05/24 06:02:49, Lasse Reichstein Nielsen wrote:
> I've rewritten this to not chain the futures, and just complete a completer at
> the end.

Working around this here before we know it's actually a problem seems like
premature optimization. So far (as far as I know) the only case where we've seen
this actually become an issue is in pathologically-exponential cases in the Pub
version solver. I don't want to add a workaround here unless we've got good
reason to believe the issue will come up in practice.

If it does come up in practice, that means that Futures are memory-hungry enough
that something like 10,000 of them can blow the heap. This seems like a serious
problem external to this code. In general, it's not unreasonable to want to
write code that iterates over a collection or stream and creates a Future for
each member. Sometimes it's unavoidable.

It's going to be untenable to try to identify every case where a Future chain
could become sufficiently long to tickle this issue, even for those of us aware
of it. For people coming to dart:async for the first time, there's no way
they'll intuit that this issue exists. We need to find a more general solution.

https://codereview.chromium.org/14753009/diff/44002/pkg/scheduled_test/lib/sr...
File pkg/scheduled_test/lib/src/mock_clock.dart (right):

https://codereview.chromium.org/14753009/diff/44002/pkg/scheduled_test/lib/sr...
pkg/scheduled_test/lib/src/mock_clock.dart:65: return controller.stream;
I'd rather see this pattern generalized, so it's clearer what's going on and can
be used elsewhere if necessary.

One possible way to do this would be to have a wrapper class like
"UncancellableBroadcastStream" that internally ensures that there's always a
dummy subscription on the wrapped stream so it doesn't get completely cancelled.

https://codereview.chromium.org/14753009/diff/44002/pkg/scheduled_test/lib/sr...
pkg/scheduled_test/lib/src/mock_clock.dart:73: void tick([int milliseconds = 1])
{
Our style is to omit spaces around "=" in default value assignment.

Alternately, omit the default value entirely and do "if (milliseconds == null)
milliseconds = 1" in the body.

https://codereview.chromium.org/14753009/diff/44002/pkg/scheduled_test/lib/sr...
File pkg/scheduled_test/lib/src/utils.dart (right):

https://codereview.chromium.org/14753009/diff/44002/pkg/scheduled_test/lib/sr...
pkg/scheduled_test/lib/src/utils.dart:101: StackTrace stackTrace = new
Trace.current();
"var stackTrace"

https://codereview.chromium.org/14753009/diff/44002/sdk/lib/_internal/pub/tes...
File sdk/lib/_internal/pub/test/error_group_test.dart (right):

https://codereview.chromium.org/14753009/diff/44002/sdk/lib/_internal/pub/tes...
sdk/lib/_internal/pub/test/error_group_test.dart:248: expect(new Future(() {
On 2013/05/24 15:53:17, floitsch wrote:
> Does this part of the test still make sense now?

It sort of does, but it has a race condition. The [controller.add] call is now
not guaranteed to run after [stream.first] completes. I think the previous
version of the test is still correct.

[floitsch, 2013-05-24 20:40:48]

https://codereview.chromium.org/14753009/diff/44002/pkg/scheduled_test/lib/sr...
File pkg/scheduled_test/lib/src/mock_clock.dart (right):

https://codereview.chromium.org/14753009/diff/44002/pkg/scheduled_test/lib/sr...
pkg/scheduled_test/lib/src/mock_clock.dart:65: return controller.stream;
On 2013/05/24 20:12:31, nweiz wrote:
> I'd rather see this pattern generalized, so it's clearer what's going on and
can
> be used elsewhere if necessary.
> 
> One possible way to do this would be to have a wrapper class like
> "UncancellableBroadcastStream" that internally ensures that there's always a
> dummy subscription on the wrapped stream so it doesn't get completely
cancelled.

https://codereview.chromium.org/16003002/

[nweiz, 2013-05-24 20:49:18]

https://codereview.chromium.org/14753009/diff/44002/pkg/scheduled_test/lib/sr...
File pkg/scheduled_test/lib/src/mock_clock.dart (right):

https://codereview.chromium.org/14753009/diff/44002/pkg/scheduled_test/lib/sr...
pkg/scheduled_test/lib/src/mock_clock.dart:65: return controller.stream;
On 2013/05/24 20:40:48, floitsch wrote:
> On 2013/05/24 20:12:31, nweiz wrote:
> > I'd rather see this pattern generalized, so it's clearer what's going on and
> can
> > be used elsewhere if necessary.
> > 
> > One possible way to do this would be to have a wrapper class like
> > "UncancellableBroadcastStream" that internally ensures that there's always a
> > dummy subscription on the wrapped stream so it doesn't get completely
> cancelled.
> 
> https://codereview.chromium.org/16003002/

That looks good. Whichever of this CL or that CL that gets submitted second
should change this code to use that constructor.

[Jennifer Messerly, 2013-05-24 23:28:08]

few more comments on second look.

https://codereview.chromium.org/14753009/diff/44002/sdk/lib/mdv_observe_impl/...
File sdk/lib/mdv_observe_impl/mdv_observe_impl.dart (right):

https://codereview.chromium.org/14753009/diff/44002/sdk/lib/mdv_observe_impl/...
sdk/lib/mdv_observe_impl/mdv_observe_impl.dart:72: _observers = new
Set<StreamController<List<ChangeRecord>>>();
fwiw... this seems awfully complicated for something that's essentially just to
exposing a simple multi-subscription event.

Is there anything we can provide to make it easier to handle this case?

Perhaps StreamController.multiplex? :)

https://codereview.chromium.org/14753009/diff/44002/sdk/lib/mdv_observe_impl/...
sdk/lib/mdv_observe_impl/mdv_observe_impl.dart:90:
List<StreamController<List<ChangeRecord>>> observers =
style nit, I'd use "var" here and below:

http://www.dartlang.org/articles/style-guide/#prefer-using-var-without-a-type...

[Lasse Reichstein Nielsen, 2013-05-27 08:04:53]

PTAL

https://codereview.chromium.org/14753009/diff/6001/pkg/scheduled_test/lib/sch...
File pkg/scheduled_test/lib/scheduled_process.dart (right):

https://codereview.chromium.org/14753009/diff/6001/pkg/scheduled_test/lib/sch...
pkg/scheduled_test/lib/scheduled_process.dart:42: StreamIterator<String>
_stdout;
On 2013/05/24 20:12:31, nweiz wrote:
> On 2013/05/24 06:02:49, Lasse Reichstein Nielsen wrote:
> > Exactly. 
> > The current code subscribed to the same stream more than once (using
> streamFirst
> > one or more time). That does not work any more.
> 
> Okay, I suppose that makes sense. It seems a little weird to me that having an
> Iterator-style API and supporting multiple subscriptions are conflated
concepts,
> but that's a different discussion and I don't feel too strongly about it.
> 
> If streamFirst is completely broken now, please remove it from utils.dart.

Done.

https://codereview.chromium.org/14753009/diff/6001/pkg/scheduled_test/lib/sch...
pkg/scheduled_test/lib/scheduled_process.dart:273: return
streamIterator.moveNext().then((bool hasNext) {
On 2013/05/24 15:54:38, Bob Nystrom wrote:
> On 2013/05/24 06:02:49, Lasse Reichstein Nielsen wrote:
> > ok.
> > Grudgingly.
> > How else would I know the type of hasNext from looking at the code? :(
> 
> 1. You know what you're passing that callback to, so you probably know the fn
> type it expects.

I don't believe in "knowing" being good enough as documentation.
Writing code is easy. Reading it is hard.

> 2. It's name starts with "has" -> probably a bool.
> 3. The type doesn't matter that much, its name tells you what it is.

You can't convince me that types are irrelevant. Redundancy is *good* when it
comes to documentation. It dispels any doubt you may have over either of the
parts in isolation.

https://codereview.chromium.org/14753009/diff/6001/pkg/scheduled_test/lib/sch...
pkg/scheduled_test/lib/scheduled_process.dart:276: return
collectAll(accumulator);
In this case, I actually think the new version is also more readable, so it's a
pure win.

In general, I doubt memory is going to be the constraining factor, but stack
space might.

We can consider special casing future-chaining, so if a future.then returns a
new future, the original future isn't just made a listener on the new future -
instead all its listeners, current and future, are moved to the new future
directly. That should reduce the nesting.

https://codereview.chromium.org/14753009/diff/44002/sdk/lib/mdv_observe_impl/...
File sdk/lib/mdv_observe_impl/mdv_observe_impl.dart (right):

https://codereview.chromium.org/14753009/diff/44002/sdk/lib/mdv_observe_impl/...
sdk/lib/mdv_observe_impl/mdv_observe_impl.dart:72: _observers = new
Set<StreamController<List<ChangeRecord>>>();
Yes, we introduced StreamController.multiplex after having had to do this in two
places. I should ofcourse go back and make those two places use it :)

https://codereview.chromium.org/14753009/diff/44002/sdk/lib/mdv_observe_impl/...
sdk/lib/mdv_observe_impl/mdv_observe_impl.dart:90:
List<StreamController<List<ChangeRecord>>> observers =
It's your code, so ok.

https://codereview.chromium.org/14753009/diff/44002/sdk/lib/mdv_observe_impl/...
sdk/lib/mdv_observe_impl/mdv_observe_impl.dart:97: // TODO(jmesserly): make
"changes" immutable
Ofcourse :)

[floitsch, 2013-05-27 08:24:43]

LGTM.

https://codereview.chromium.org/14753009/diff/68001/sdk/lib/async/stream_impl...
File sdk/lib/async/stream_impl.dart (right):

https://codereview.chromium.org/14753009/diff/68001/sdk/lib/async/stream_impl...
sdk/lib/async/stream_impl.dart:399: * After ing, no further callbacks will
happen.
bad search/replace.

https://codereview.chromium.org/14753009/diff/68001/sdk/lib/async/stream_impl...
sdk/lib/async/stream_impl.dart:1084: // If in on of the _STATE_EXTRA_* states,
the _cache object may hold the
When in one...

https://codereview.chromium.org/14753009/diff/68001/sdk/lib/async/stream_impl...
sdk/lib/async/stream_impl.dart:1133: Future<bool> moveNext() {
comment.

https://codereview.chromium.org/14753009/diff/68001/sdk/lib/async/stream_impl...
sdk/lib/async/stream_impl.dart:1164: void _clear() {
Comment.

https://codereview.chromium.org/14753009/diff/68001/sdk/lib/async/stream_impl...
sdk/lib/async/stream_impl.dart:1171: void cancel() {
comment.

https://codereview.chromium.org/14753009/diff/68001/sdk/lib/async/stream_impl...
sdk/lib/async/stream_impl.dart:1193: _futureOrPrefetch = data;
assert that _futureOrPrefetch == null

https://codereview.chromium.org/14753009/diff/77001/sdk/lib/async/stream_impl...
File sdk/lib/async/stream_impl.dart (right):

https://codereview.chromium.org/14753009/diff/77001/sdk/lib/async/stream_impl...
sdk/lib/async/stream_impl.dart:1079: // The interpretation of the
[_futureOrPrefecth] field depends on the state.
Prefetch

[Lasse Reichstein Nielsen, 2013-05-27 08:33:44]

https://codereview.chromium.org/14753009/diff/68001/sdk/lib/async/stream_impl...
File sdk/lib/async/stream_impl.dart (right):

https://codereview.chromium.org/14753009/diff/68001/sdk/lib/async/stream_impl...
sdk/lib/async/stream_impl.dart:399: * After ing, no further callbacks will
happen.
On 2013/05/27 08:24:43, floitsch wrote:
> bad search/replace.

Done.

https://codereview.chromium.org/14753009/diff/68001/sdk/lib/async/stream_impl...
sdk/lib/async/stream_impl.dart:1084: // If in on of the _STATE_EXTRA_* states,
the _cache object may hold the
On 2013/05/27 08:24:43, floitsch wrote:
> When in one...

Done.

https://codereview.chromium.org/14753009/diff/68001/sdk/lib/async/stream_impl...
sdk/lib/async/stream_impl.dart:1133: Future<bool> moveNext() {
In the interface.

https://codereview.chromium.org/14753009/diff/68001/sdk/lib/async/stream_impl...
sdk/lib/async/stream_impl.dart:1164: void _clear() {
On 2013/05/27 08:24:43, floitsch wrote:
> Comment.

Done.

https://codereview.chromium.org/14753009/diff/68001/sdk/lib/async/stream_impl...
sdk/lib/async/stream_impl.dart:1171: void cancel() {
In the interface.

https://codereview.chromium.org/14753009/diff/68001/sdk/lib/async/stream_impl...
sdk/lib/async/stream_impl.dart:1193: _futureOrPrefetch = data;
On 2013/05/27 08:24:43, floitsch wrote:
> assert that _futureOrPrefetch == null

Done.

[Lasse Reichstein Nielsen, 2013-05-27 08:34:38]

Committed patchset #15 manually as r23204 (presubmit successful).