Make StreamController be a StreamSink, not just an EventSink.

Make StreamController be a StreamSink, not just an EventSink.

BUG=http://dartbug.com/10677
R=floitsch@google.com

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

[floitsch, 2013-06-06 15:08:29]

LGTM once test is addressed.
Also: should we make a StreamController implement StreamSink?

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

https://codereview.chromium.org/16240008/diff/1/sdk/lib/async/stream_controll...
sdk/lib/async/stream_controller.dart:315: _runGuarded(_onPause);
Do we want to call onPause when there is an addStream active?

https://codereview.chromium.org/16240008/diff/1/sdk/lib/async/stream_controll...
sdk/lib/async/stream_controller.dart:320: _runGuarded(_onResume);
ditto.

https://codereview.chromium.org/16240008/diff/1/sdk/lib/async/stream_controll...
sdk/lib/async/stream_controller.dart:509: // Cached return value of [sink]. Used
to cancel a `sink.addStream`
cancel, pause and resume a `sink.addStream`.

https://codereview.chromium.org/16240008/diff/1/sdk/lib/async/stream_controll...
sdk/lib/async/stream_controller.dart:658: if (_sink != null && isClosed) {
add comment explaining why you look at `isClosed`.

https://codereview.chromium.org/16240008/diff/1/tests/lib/async/stream_contro...
File tests/lib/async/stream_controller_async_test.dart (right):

https://codereview.chromium.org/16240008/diff/1/tests/lib/async/stream_contro...
tests/lib/async/stream_controller_async_test.dart:560: // This is wrong, it
should get more events before being done.
Don't understand.
The subscription is only paused. So the other elements (4, 5, 43) should be in
the expected list. no?

https://codereview.chromium.org/16240008/diff/1/tests/lib/async/stream_contro...
tests/lib/async/stream_controller_async_test.dart:566: sub.pause(new
Future.delayed(const Duration(milliseconds: 15), () {}));
long line.

[Lasse Reichstein Nielsen, 2013-06-10 13:52:56]

I think it would make great sense to make the constroller itself a StreamSink.
Let's do that instead.
Then the .sink wrapper will again just be a wrapper.

[Lasse Reichstein Nielsen, 2013-06-27 07:46:52]

Complete rewrite, PTAL.

[floitsch, 2013-06-27 15:15:18]

LGTM, but I think that an addStream on a stream-controller should never complete
with an error. Errors should be transmitted to the actual stream.

Also: I think that broadcast-streams should cancel running streams, when they
don't have any listener anymore. But that's up to discussion.

https://codereview.chromium.org/16240008/diff/7001/sdk/lib/async/broadcast_st...
File sdk/lib/async/broadcast_stream_controller.dart (right):

https://codereview.chromium.org/16240008/diff/7001/sdk/lib/async/broadcast_st...
sdk/lib/async/broadcast_stream_controller.dart:51: bool _expectsEvent(int
eventId) {
=> ?

https://codereview.chromium.org/16240008/diff/7001/sdk/lib/async/broadcast_st...
sdk/lib/async/broadcast_stream_controller.dart:55: void _toggleEventId() {
=> ?

https://codereview.chromium.org/16240008/diff/7001/sdk/lib/async/broadcast_st...
sdk/lib/async/broadcast_stream_controller.dart:66: bool get _removeAfterFiring
=>
_shouldRemoveAfterFiring

https://codereview.chromium.org/16240008/diff/7001/sdk/lib/async/broadcast_st...
sdk/lib/async/broadcast_stream_controller.dart:72: }
missing _onCancel() from the _ControllerSubscription.

https://codereview.chromium.org/16240008/diff/7001/sdk/lib/async/broadcast_st...
sdk/lib/async/broadcast_stream_controller.dart:105: * This means when all
listeners at the time when the done event was
bad English sentence.

https://codereview.chromium.org/16240008/diff/7001/sdk/lib/async/broadcast_st...
sdk/lib/async/broadcast_stream_controller.dart:159: previous._next =
subscription;
needs comments.
Either explain that you want to accept cyclic lists, or don't do magic things
with subscription._previous.

https://codereview.chromium.org/16240008/diff/7001/sdk/lib/async/broadcast_st...
sdk/lib/async/broadcast_stream_controller.dart:169: subscription._previous._next
= subscription._next;
please make this nicer to read:
var prev = sub.prev;
var next = sub.next;
prev.next = next;
next.prev = prev;
sub.next = sub.prev = sub;

https://codereview.chromium.org/16240008/diff/7001/sdk/lib/async/broadcast_st...
sdk/lib/async/broadcast_stream_controller.dart:183: return new
_DoneSubscription<T>();
Let's throw instead.

https://codereview.chromium.org/16240008/diff/7001/sdk/lib/async/broadcast_st...
sdk/lib/async/broadcast_stream_controller.dart:206: if ((_state & _STATE_FIRING)
== 0 && _isEmpty) {
if (!_isFiring && _isEmpty)

https://codereview.chromium.org/16240008/diff/7001/sdk/lib/async/broadcast_st...
sdk/lib/async/broadcast_stream_controller.dart:257: // _EventSink interface,
called from AddStramState.
AddStreamState

https://codereview.chromium.org/16240008/diff/7001/sdk/lib/async/broadcast_st...
sdk/lib/async/broadcast_stream_controller.dart:263: assert(_isAddingStream);
Why is an error fatal? Isn't it just passed through?

https://codereview.chromium.org/16240008/diff/7001/sdk/lib/async/future_impl....
File sdk/lib/async/future_impl.dart (right):

https://codereview.chromium.org/16240008/diff/7001/sdk/lib/async/future_impl....
sdk/lib/async/future_impl.dart:318: void _setErrorUnchecked(error) {
If you add the "Object" above, also add it here.

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

https://codereview.chromium.org/16240008/diff/7001/sdk/lib/async/stream_contr...
sdk/lib/async/stream_controller.dart:140: * If true, the "done" event might not
have _ yet, but it has been
undo change?

https://codereview.chromium.org/16240008/diff/7001/sdk/lib/async/stream_contr...
sdk/lib/async/stream_controller.dart:211: // If executing an [addStream], now
events are not allowed either, but will
-now-

Move description of _STATE_ADDSTREAM below. Otherwise it is not clear if the
comment applies to _STATE_CLOSED too. like:
"Does that mean that the flag is set when doing an [addStream]?"

https://codereview.chromium.org/16240008/diff/7001/sdk/lib/async/stream_contr...
sdk/lib/async/stream_controller.dart:421: // Error from addStream. Stop the
addStream and complete its future with the
I don't think that's the right thing to do. Just pass it through.

https://codereview.chromium.org/16240008/diff/7001/sdk/lib/async/stream_contr...
sdk/lib/async/stream_controller.dart:632: StreamSink _target;
final.

https://codereview.chromium.org/16240008/diff/7001/tests/lib/async/stream_con...
File tests/lib/async/stream_controller_async_test.dart (right):

https://codereview.chromium.org/16240008/diff/7001/tests/lib/async/stream_con...
tests/lib/async/stream_controller_async_test.dart:548: // stream, it will still
be open, and we read the done" future before
spurious '"'.

https://codereview.chromium.org/16240008/diff/7001/tests/lib/async/stream_con...
tests/lib/async/stream_controller_async_test.dart:572:
expected..add(4)..add(5)..add(43)..close();
If I understand correctly the broadcast subscription will buffer on its own for
15ms and then only add the remaining events.
So if we waited before doing the Expect.listEquals we would get the same result
on both sides?
If yes we should test it:

var delayedFuture;

in sub.pause: delayedFuture = ...;

in the final .then (where we execute the listEquals): check twice: once before,
and once after the delayedFuture is done.

https://codereview.chromium.org/16240008/diff/7001/tests/lib/async/stream_con...
tests/lib/async/stream_controller_async_test.dart:581: () => null));
bad indentation

https://codereview.chromium.org/16240008/diff/7001/tests/lib/async/stream_con...
tests/lib/async/stream_controller_async_test.dart:614: testSink(sync: false,
broadcast: true, asBroadcast: false);
add test, when the addStream produces errors.

[floitsch, 2013-06-27 16:27:00]

Also: please run the dart-analyzer (or open in the editor), if you haven't done
so.

[Lasse Reichstein Nielsen, 2013-06-28 12:56:45]

Committed patchset #3 manually as r24587 (presubmit successful).

[Lasse Reichstein Nielsen, 2013-06-28 12:57:38]

https://codereview.chromium.org/16240008/diff/7001/sdk/lib/async/broadcast_st...
File sdk/lib/async/broadcast_stream_controller.dart (right):

https://codereview.chromium.org/16240008/diff/7001/sdk/lib/async/broadcast_st...
sdk/lib/async/broadcast_stream_controller.dart:51: bool _expectsEvent(int
eventId) {
Will do. 
Probably added the block to be able to do more during debugging.

https://codereview.chromium.org/16240008/diff/7001/sdk/lib/async/broadcast_st...
sdk/lib/async/broadcast_stream_controller.dart:55: void _toggleEventId() {
doesn't return a value.

https://codereview.chromium.org/16240008/diff/7001/sdk/lib/async/broadcast_st...
sdk/lib/async/broadcast_stream_controller.dart:66: bool get _removeAfterFiring
=>
This is an imperative. It must be removed after firing. And it WILL be removed
after firing. There is no "should" about it.

https://codereview.chromium.org/16240008/diff/7001/sdk/lib/async/broadcast_st...
sdk/lib/async/broadcast_stream_controller.dart:72: }
It's inherited. We overwrite _onPause and _onResume because we know they will do
nothing, so there is no need to call the controller's _recordPause/Resume to do
nothing there.

Added comments.

https://codereview.chromium.org/16240008/diff/7001/sdk/lib/async/broadcast_st...
sdk/lib/async/broadcast_stream_controller.dart:105: * This means when all
listeners at the time when the done event was
Reworded.

https://codereview.chromium.org/16240008/diff/7001/sdk/lib/async/broadcast_st...
sdk/lib/async/broadcast_stream_controller.dart:159: previous._next =
subscription;
On 2013/06/27 15:15:19, floitsch wrote:
> needs comments.
> Either explain that you want to accept cyclic lists, or don't do magic things
> with subscription._previous.

Done.

https://codereview.chromium.org/16240008/diff/7001/sdk/lib/async/broadcast_st...
sdk/lib/async/broadcast_stream_controller.dart:169: subscription._previous._next
= subscription._next;
On 2013/06/27 15:15:19, floitsch wrote:
> please make this nicer to read:
> var prev = sub.prev;
> var next = sub.next;
> prev.next = next;
> next.prev = prev;
> sub.next = sub.prev = sub;

Done.

https://codereview.chromium.org/16240008/diff/7001/sdk/lib/async/broadcast_st...
sdk/lib/async/broadcast_stream_controller.dart:183: return new
_DoneSubscription<T>();
On 2013/06/27 15:15:19, floitsch wrote:
> Let's throw instead.

Done.

https://codereview.chromium.org/16240008/diff/7001/sdk/lib/async/broadcast_st...
sdk/lib/async/broadcast_stream_controller.dart:206: if ((_state & _STATE_FIRING)
== 0 && _isEmpty) {
On 2013/06/27 15:15:19, floitsch wrote:
> if (!_isFiring && _isEmpty)

Done.

https://codereview.chromium.org/16240008/diff/7001/sdk/lib/async/broadcast_st...
sdk/lib/async/broadcast_stream_controller.dart:257: // _EventSink interface,
called from AddStramState.
On 2013/06/27 15:15:19, floitsch wrote:
> AddStreamState

Done.

https://codereview.chromium.org/16240008/diff/7001/sdk/lib/async/broadcast_st...
sdk/lib/async/broadcast_stream_controller.dart:263: assert(_isAddingStream);
Let's pass it through, the controller can handle it.

https://codereview.chromium.org/16240008/diff/7001/sdk/lib/async/future_impl....
File sdk/lib/async/future_impl.dart (right):

https://codereview.chromium.org/16240008/diff/7001/sdk/lib/async/future_impl....
sdk/lib/async/future_impl.dart:318: void _setErrorUnchecked(error) {
On 2013/06/27 15:15:19, floitsch wrote:
> If you add the "Object" above, also add it here.

Done.

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

https://codereview.chromium.org/16240008/diff/7001/sdk/lib/async/stream_contr...
sdk/lib/async/stream_controller.dart:140: * If true, the "done" event might not
have _ yet, but it has been
On 2013/06/27 15:15:19, floitsch wrote:
> undo change?

Done.

https://codereview.chromium.org/16240008/diff/7001/sdk/lib/async/stream_contr...
sdk/lib/async/stream_controller.dart:211: // If executing an [addStream], now
events are not allowed either, but will
now -> new.
Reworded and reordered.

https://codereview.chromium.org/16240008/diff/7001/sdk/lib/async/stream_contr...
sdk/lib/async/stream_controller.dart:421: // Error from addStream. Stop the
addStream and complete its future with the
On 2013/06/27 15:15:19, floitsch wrote:
> I don't think that's the right thing to do. Just pass it through.

Done.

https://codereview.chromium.org/16240008/diff/7001/sdk/lib/async/stream_contr...
sdk/lib/async/stream_controller.dart:632: StreamSink _target;
On 2013/06/27 15:15:19, floitsch wrote:
> final.

Done.

https://codereview.chromium.org/16240008/diff/7001/tests/lib/async/stream_con...
File tests/lib/async/stream_controller_async_test.dart (right):

https://codereview.chromium.org/16240008/diff/7001/tests/lib/async/stream_con...
tests/lib/async/stream_controller_async_test.dart:548: // stream, it will still
be open, and we read the done" future before
On 2013/06/27 15:15:19, floitsch wrote:
> spurious '"'.

Done.

https://codereview.chromium.org/16240008/diff/7001/tests/lib/async/stream_con...
tests/lib/async/stream_controller_async_test.dart:581: () => null));
On 2013/06/27 15:15:19, floitsch wrote:
> bad indentation

Done.

https://codereview.chromium.org/16240008/diff/7001/tests/lib/async/stream_con...
tests/lib/async/stream_controller_async_test.dart:595: .then((_) {
I'll try adding a pause here (50 ms, just to be sure we're past the 15ms pause).