Add groupBy to Stream.

Add groupBy to Stream.

R=floitsch@google.com

Committed: https://github.com/dart-lang/sdk/commit/3e8bfb1adbf7ee156644ed0d4327f410c91a6564

[Lasse Reichstein Nielsen, 2017-05-02 10:20:14]

lrn@google.com changed reviewers:
+ floitsch@google.com

[floitsch, 2017-05-02 12:23:53]

LGTM.

Requires a changelog entry.

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

https://codereview.chromium.org/2850393003/diff/20001/sdk/lib/async/stream.da...
sdk/lib/async/stream.dart:396: /// output [StreamGroup.values] stream.
That sentence sounds wrong.

https://codereview.chromium.org/2850393003/diff/20001/sdk/lib/async/stream.da...
sdk/lib/async/stream.dart:409: /// is paused and until it is first listened to.
Mention that this could lead to memory issues.

Also: what happens, if I only listen to one of the streams? I guess it leads to
memory issues at well.

https://codereview.chromium.org/2850393003/diff/20001/sdk/lib/async/stream.da...
sdk/lib/async/stream.dart:428: /// of the `key` function, is reported on the
stream for the corresponding key.
long line.

https://codereview.chromium.org/2850393003/diff/20001/sdk/lib/async/stream.da...
sdk/lib/async/stream.dart:439: Stream<StreamGroup<K, V>> groupValuesBy<K, V>(
I think it would be cleaner if `StreamGroup` had a `mapValues` function.

stream
    .groupBy(...)
    .map((group) => group.mapValues(...))

There could be other methods that could be similarly interesting. For example a
`map` function that also takes the `key` as argument. (Not sure, if it should
become a Stream then, or stay a StreamGroup).

If we keep this method, I would prefer `groupByMapped`.

https://codereview.chromium.org/2850393003/diff/20001/sdk/lib/async/stream.da...
sdk/lib/async/stream.dart:446: void closeAll() {
New line before nested function.

https://codereview.chromium.org/2850393003/diff/20001/sdk/lib/async/stream.da...
sdk/lib/async/stream.dart:453: (data) {
This looks like it wasn't dartfmt.

https://codereview.chromium.org/2850393003/diff/20001/sdk/lib/async/stream.da...
sdk/lib/async/stream.dart:472: groupController.addError(error, stackTrace);
Since the group controller is sync, is there enough time to register for the
error handler?

Off topic: small tests show, that we simply buffer errors on streams until there
is a listener. That seems to indicate that we should allow `sync: true` much
more often.

https://codereview.chromium.org/2850393003/diff/20001/sdk/lib/async/stream.da...
sdk/lib/async/stream.dart:1909: class StreamGroup<K, V> {
This class could have:

StreamGroup<K, V2> mapValues<V2>(V2 Function(V) f);
StreamGroup<K, V2> map<V2>(V2 Function(K, V) f); or
Stream<V2> map<V2>(V2 Function(K, V) f);

Future close();  // Since the stream would otherwise buffer.
or
Future cancel();

https://codereview.chromium.org/2850393003/diff/20001/tests/lib/async/stream_...
File tests/lib/async/stream_group_by_test.dart (right):

https://codereview.chromium.org/2850393003/diff/20001/tests/lib/async/stream_...
tests/lib/async/stream_group_by_test.dart:69: // groupValuesBy
Nit. finish with ".".

https://codereview.chromium.org/2850393003/diff/20001/tests/lib/async/stream_...
tests/lib/async/stream_group_by_test.dart:70: test("splits", () async {
There is already a test called "splits".

[Lasse Reichstein Nielsen, 2017-05-04 11:05:20]

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

https://codereview.chromium.org/2850393003/diff/20001/sdk/lib/async/stream.da...
sdk/lib/async/stream.dart:396: /// output [StreamGroup.values] stream.
Rewritten.

https://codereview.chromium.org/2850393003/diff/20001/sdk/lib/async/stream.da...
sdk/lib/async/stream.dart:409: /// is paused and until it is first listened to.
Added something about memory issues.

https://codereview.chromium.org/2850393003/diff/20001/sdk/lib/async/stream.da...
sdk/lib/async/stream.dart:428: /// of the `key` function, is reported on the
stream for the corresponding key.
On 2017/05/02 12:23:53, floitsch wrote:
> long line.

Done.

https://codereview.chromium.org/2850393003/diff/20001/sdk/lib/async/stream.da...
sdk/lib/async/stream.dart:439: Stream<StreamGroup<K, V>> groupValuesBy<K, V>(
On 2017/05/02 12:23:53, floitsch wrote:
> I think it would be cleaner if `StreamGroup` had a `mapValues` function.

That is probably more interesting. But then the next question is why it doesn't
have all the other Stream methods, like `where` or `expand`.

If only we had simple tuples, then a group would just be (key, values)
and you could just do: `s.groupBy(...).map((key,values)=>(key,values.map(...)))`

I don't think this scales, so I'd prefer to not do it.
Let's just not do anything for now, not even groupByMapped, and see what users
actually ask for.

https://codereview.chromium.org/2850393003/diff/20001/sdk/lib/async/stream.da...
sdk/lib/async/stream.dart:446: void closeAll() {
On 2017/05/02 12:23:53, floitsch wrote:
> New line before nested function.

Done.

https://codereview.chromium.org/2850393003/diff/20001/sdk/lib/async/stream.da...
sdk/lib/async/stream.dart:453: (data) {
Looks deceive :)

https://codereview.chromium.org/2850393003/diff/20001/sdk/lib/async/stream.da...
sdk/lib/async/stream.dart:472: groupController.addError(error, stackTrace);
That's expected behavior - adding errors to a paused stream, even a sync one, is
safe, they will be delivered asynchronously after a resume or listen.

It's sending events *during* onListen that's the problem, not sending them
before. They won't be unhandled until someone actually listens.

https://codereview.chromium.org/2850393003/diff/20001/sdk/lib/async/stream.da...
sdk/lib/async/stream.dart:1909: class StreamGroup<K, V> {
On 2017/05/02 12:23:53, floitsch wrote:
> This class could have:
> 
> StreamGroup<K, V2> mapValues<V2>(V2 Function(V) f);
> StreamGroup<K, V2> map<V2>(V2 Function(K, V) f); or
> Stream<V2> map<V2>(V2 Function(K, V) f);

As said above, let's not do that (yet).


> Future close();  // Since the stream would otherwise buffer.
> or
> Future cancel();

Using "cancel".

https://codereview.chromium.org/2850393003/diff/20001/tests/lib/async/stream_...
File tests/lib/async/stream_group_by_test.dart (right):

https://codereview.chromium.org/2850393003/diff/20001/tests/lib/async/stream_...
tests/lib/async/stream_group_by_test.dart:69: // groupValuesBy
Every test below here removed. Problem solved!

https://codereview.chromium.org/2850393003/diff/20001/tests/lib/async/stream_...
tests/lib/async/stream_group_by_test.dart:70: test("splits", () async {
On 2017/05/02 12:23:53, floitsch wrote:
> There is already a test called "splits".

Acknowledged.

[Lasse Reichstein Nielsen, 2017-05-04 11:09:30]

PTAL. Will add changelog entry.

[floitsch, 2017-05-04 13:32:09]

LGTM.

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

https://codereview.chromium.org/2850393003/diff/40001/sdk/lib/async/stream.da...
sdk/lib/async/stream.dart:410: /// it will enqueue all the events unnecessarily.
One should therefore not ignore streams of keys that aren't used, but rather
[StreamGroup.cancel] the group.

[Lasse Reichstein Nielsen, 2017-05-08 08:25:21]

Description was changed from

==========
Add groupBy to Stream.
==========

to

==========
Add groupBy to Stream.

R=floitsch@google.com

Committed:
https://github.com/dart-lang/sdk/commit/3e8bfb1adbf7ee156644ed0d4327f410c91a6564
==========

[Lasse Reichstein Nielsen, 2017-05-08 08:25:24]

Committed patchset #4 (id:60001) manually as
3e8bfb1adbf7ee156644ed0d4327f410c91a6564 (presubmit successful).