Change and structure how Stream implementations do callbacks.

Change and structure how Stream implementations do callbacks.

Callbacks are the on-pause/subscription-state change calls from a stream.

Now a callback is an atomic operation. State changes happening during a callback
will not trigger a new callback before the old one is done. If the state is
changed during the callback, another callback will follow immediately.

Firing of events is also atomic, so a callback will not happend during an event
being fired, but will be triggered afterwards if the state was changed by the
event handlers.

A stream controller (or any other listener for callbacks) will not be told
immediately if a paused stream is resumed. If there are pending events in the
queue, those will be fired before the controller is considerered resumed.

A canceled subscription will immediately stop getting events.
Previously it would get the current event being fired, but it's more rational
to not get events after you have said you don't want them, instead of having
the handlers have to discard events after a cancel.

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

[floitsch, 2013-03-01 21:52:30]

LGTM.
But the Stream implementation is still way too complex.
Let's discuss on Monday if we can find ways to make it simpler.

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

https://codereview.chromium.org/12393009/diff/1/sdk/lib/async/stream.dart#new...
sdk/lib/async/stream.dart:25: * is canceled, it immediately stops receiveing
events.
receiving

https://codereview.chromium.org/12393009/diff/1/sdk/lib/async/stream_controll...
File sdk/lib/async/stream_controller.dart (right):

https://codereview.chromium.org/12393009/diff/1/sdk/lib/async/stream_controll...
sdk/lib/async/stream_controller.dart:23: * If the stream starts or stops having
listeners (first listener subscribing,
We could make our lives easier by making the StreamController
single-subscription.
Users could use "asBroadcast" if they really want multiple subscribers.

https://codereview.chromium.org/12393009/diff/1/sdk/lib/async/stream_controll...
sdk/lib/async/stream_controller.dart:27: * If the pause state has also changed,
only the subscription state callback
Not clear: are you saying that a subscription change will hide a pause-change?

so stream.listen(...).pause() will not call the pause-change?

The way I read it, this is not the case, since both the 'listen' and the 'pause'
would be immediate, but it is not completely clear.

https://codereview.chromium.org/12393009/diff/1/sdk/lib/async/stream_controll...
sdk/lib/async/stream_controller.dart:32: * a current event or completing a
another callback. This happens if the stream
-a-

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

https://codereview.chromium.org/12393009/diff/1/sdk/lib/async/stream_impl.dar...
sdk/lib/async/stream_impl.dart:168: // State prediates.
predicates

https://codereview.chromium.org/12393009/diff/1/sdk/lib/async/stream_impl.dar...
sdk/lib/async/stream_impl.dart:320: bool wasControllerPaused =
_isControllerPaused;
what about calling this "wasInputPaused" ?

https://codereview.chromium.org/12393009/diff/1/sdk/lib/async/stream_impl.dar...
sdk/lib/async/stream_impl.dart:346: // Controller's pause state haven't changed.
hasn't

https://codereview.chromium.org/12393009/diff/1/sdk/lib/async/stream_impl.dar...
sdk/lib/async/stream_impl.dart:462: * cases events may be fired from here.
That sounds bad. adding a pending event should not trigger fire it.

_add, _signalError and _close already call _handlePendingEvents() if they are
allowed to fire.

https://codereview.chromium.org/12393009/diff/1/sdk/lib/async/stream_impl.dar...
sdk/lib/async/stream_impl.dart:575: * * [_isControllerPaused]: Test whether the
stream is currently paused.
We still have the "_isPaused" too. Is that one not considered "public" anymore?

https://codereview.chromium.org/12393009/diff/1/sdk/lib/async/stream_impl.dar...
sdk/lib/async/stream_impl.dart:640: // Keep being paused while there is no
subscriber and the stream is not
I don't understand this:
why do we pause a Stream without a subscriber?
when do we call the subscriptionStateChange?

https://codereview.chromium.org/12393009/diff/1/tests/lib/async/stream_state_...
File tests/lib/async/stream_state_test.dart (right):

https://codereview.chromium.org/12393009/diff/1/tests/lib/async/stream_state_...
tests/lib/async/stream_state_test.dart:120: ..expectSubscription(false,
!broadcast);
why is the non-broadcast one paused?
It should be as dead as the broadcast one.

https://codereview.chromium.org/12393009/diff/1/tests/lib/async/stream_state_...
tests/lib/async/stream_state_test.dart:182: t.subscribe();
ok. That's new behavior.

[Lasse Reichstein Nielsen, 2013-03-04 11:53:02]

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

https://codereview.chromium.org/12393009/diff/1/sdk/lib/async/stream.dart#new...
sdk/lib/async/stream.dart:25: * is canceled, it immediately stops receiveing
events.
On 2013/03/01 21:52:30, floitsch wrote:
> receiving

Done.

https://codereview.chromium.org/12393009/diff/1/sdk/lib/async/stream_controll...
File sdk/lib/async/stream_controller.dart (right):

https://codereview.chromium.org/12393009/diff/1/sdk/lib/async/stream_controll...
sdk/lib/async/stream_controller.dart:23: * If the stream starts or stops having
listeners (first listener subscribing,
An asBroadCast single-subscription stream is not the same as a real broadcast
stream. The asBroadCast-single-stream is paused, like a single-subscription
stream, until the first subscriber arrives, where a real broadcast stream is
unpaused when it's created.

https://codereview.chromium.org/12393009/diff/1/sdk/lib/async/stream_controll...
sdk/lib/async/stream_controller.dart:27: * If the pause state has also changed,
only the subscription state callback
Correct, if during the same callback/event, both subscription and pause is
changed, only a subscription event is passed. At any other time, the callback
comes right after the action that changed it.

> 
> so stream.listen(...).pause() will not call the pause-change?

It will call.

> 
> The way I read it, this is not the case, since both the 'listen' and the
'pause'
> would be immediate, but it is not completely clear.

Reading is correct. Reworded.

https://codereview.chromium.org/12393009/diff/1/sdk/lib/async/stream_controll...
sdk/lib/async/stream_controller.dart:32: * a current event or completing a
another callback. This happens if the stream
On 2013/03/01 21:52:30, floitsch wrote:
> -a-

Done.

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

https://codereview.chromium.org/12393009/diff/1/sdk/lib/async/stream_impl.dar...
sdk/lib/async/stream_impl.dart:168: // State prediates.
On 2013/03/01 21:52:30, floitsch wrote:
> predicates

Done.

https://codereview.chromium.org/12393009/diff/1/sdk/lib/async/stream_impl.dar...
sdk/lib/async/stream_impl.dart:320: bool wasControllerPaused =
_isControllerPaused;
The real name should be "wasLastPauseStateChangeCallbackAPause".
The controller itself isn't paused, it can do whatever it wants, but it thinks
the stream is paused (even though the stream might already be sending pending
out events).
But "Input" is less specific, I'll change it to that.

https://codereview.chromium.org/12393009/diff/1/sdk/lib/async/stream_impl.dar...
sdk/lib/async/stream_impl.dart:462: * cases events may be fired from here.
True, it is unnecessary here.

https://codereview.chromium.org/12393009/diff/1/sdk/lib/async/stream_impl.dar...
sdk/lib/async/stream_impl.dart:575: * * [_isControllerPaused]: Test whether the
stream is currently paused.
True, it's not considered "public".
Only use it internally if you want to know if sending events is blocked.

https://codereview.chromium.org/12393009/diff/1/sdk/lib/async/stream_impl.dar...
sdk/lib/async/stream_impl.dart:640: // Keep being paused while there is no
subscriber and the stream is not
This is how single-subscription streams have worked so far.
Previously it was hacked in by changing the _isPaused predicate. I'm making it
explicit.

The subscriber callback is called by _checkCallbacks.

https://codereview.chromium.org/12393009/diff/1/tests/lib/async/stream_state_...
File tests/lib/async/stream_state_test.dart (right):

https://codereview.chromium.org/12393009/diff/1/tests/lib/async/stream_state_...
tests/lib/async/stream_state_test.dart:120: ..expectSubscription(false,
!broadcast);
Because that's how single-subscription streams work.
It might not have been the intent, but that's how they have worked so far.

https://codereview.chromium.org/12393009/diff/1/tests/lib/async/stream_state_...
tests/lib/async/stream_state_test.dart:182: t.subscribe();
Depends on what you are referring to.
It is new that we don't send any subscription-change events, but from the
controller's perspective there is no reason: before we sent the event, we had a
subscriber, after we do too.

It was possible to do this before as well, it just fired more events. Try
running this short example:

import "dart:async";

main() {
  var c = new StreamController();
  var s;
  s = c.stream.listen((v) {
     print("1:$v");
     s.cancel();
     c.stream.listen((v) { print("2:$v"); });
  });
  c.add(42);
  c.add(43);
  c.close();
}

It prints "1:42" and "2:43", so resubscribing works.
We never implemented the "single-subscriber-is-dead-when-unsubscribed" thing.

[Lasse Reichstein Nielsen, 2013-03-04 11:53:24]

Committed patchset #2 manually as r19379 (presubmit successful).