Add `StreamExtensions.buffer()` #235

nex3 · 2023-02-08T01:54:05Z

The immediate motivation for this is to buffer stderr from a subprocess until we learn whether the subprocess completed successfully or not. It's important to consume the output eagerly so the process doesn't deadlock after saturating the OS and/or dart:io buffers, but we don't want to print it at all if the process completes successfully.

I think it's generally useful to have an explicit way of doing this, since otherwise users may naively try to write Stream.pipe(StreamController) which doesn't actually buffer output if the controller doesn't have a listener.

natebosch

@lrhn - do you think this would fit better in stream_transform while it's still separate from async?

natebosch · 2023-02-17T00:37:44Z

lib/src/stream_extensions.dart

+
+  /// Returns a view of this stream that eagerly buffers all events until
+  /// `listen()` is called.
+  Stream<T> buffer() {


This name conflicts with buffer from stream_transform.

https://pub.dev/documentation/stream_transform/latest/stream_transform/RateLimit/buffer.html

WDYT about asSingleSubscriptionStream to contrast with asBroadcastStream?

Would it be sensible to also have

if (!isBroadcast) return this;

Would it be sensible to also have

I guess if the goal is specifically to have a listen on streams that might not otherwise ever have one it wouldn't make sense to return this.

We could also consider the name eagerListen instead of asSingleSubscriptionStream to reinforce the intent.

I think eagerListen() or bufferUntilListen() would make sense as names.

Or eagerlyBuffer().
The bufferUntilListen seems most descriptive, eagerListen sounds like it's more eager than just doing listen now, which isn't possible.

I like bufferUntilListen even if it's a bit verbose.

The listenAndBuffer is technically correct. It's a low-level (low abstraction) name because it describes the implementation, not a higher-level concept.
Maybe because there is no higher level concept, this really is a shorthand for listening-and-buffering, which will only be used as a low-level primitive of other things. Users shouldn't need to use it directly.

@lrhn - is that a vote for listenAndBuffer or do you still prefer bufferUntilListen?

I do think listenAndBuffer is the least likely to be misread.

Let's say it's a vote for listenAndBuffer. It's an ugly name, but it's precise and understandable.

I like that listen is early, because listening to a stream is a significant action to take. You'd usually not expect a function returning a stream to do anything before you listen to that stream.

If the name is missing something, it is what it's behavior is after listening to the returned stream.
But adding more would be too long, so putting that in the documentation is fine.

👍

@nex3 - can you rename to listenAndBuffer? Also update the name in the CHANGELOG.

Oops, done!

natebosch · 2023-02-17T00:45:01Z

lib/src/stream_extensions.dart

+        onPause: () => subscription.pause(),
+        onResume: () => subscription.resume(),
+        onCancel: () => subscription.cancel());
+    subscription = listen(controller.add,


I think it is safe to pass cancelOnError: false here.

If the result is listened with cancelOnError: false, we'd need it here too or that parameter won't work.

If the result is listened with cancelOnError: true, it should be OK. We'll add multiple errors here, but close the subscription once the result's listener closes on the first error.

I agree—but isn't cancelOnError: false the default?

🤦 - it is the default.

natebosch · 2023-02-17T00:48:39Z

It's important to consume the output eagerly so the process doesn't deadlock after saturating the OS and/or dart:io buffers

We don't get unlimited extra buffering here right? There is still backpressure through pausing the subscription, so is it still possible to fill up an OS buffer?

nex3 · 2023-02-17T01:32:47Z

We don't get unlimited extra buffering here right? There is still backpressure through pausing the subscription, so is it still possible to fill up an OS buffer?

That's right, it buffers until the point listen() is called and then behaves like a normal stream with normal backpressure. We could potentially add a bufferWhilePaused parameter as well, either now or in the future.

lrhn

Seems reasonable, implementation can be simplified. Feature can potentially be made more feature-full, but that can always be added later if there is an actual need.

lib/src/stream_extensions.dart

lrhn · 2023-02-17T10:25:11Z

lib/src/stream_extensions.dart

+
+  /// Returns a view of this stream that eagerly buffers all events until
+  /// `listen()` is called.
+  Stream<T> buffer() {


Or eagerlyBuffer().
The bufferUntilListen seems most descriptive, eagerListen sounds like it's more eager than just doing listen now, which isn't possible.

lrhn · 2023-02-17T10:29:59Z

I wouldn't put anything new into stream_transformers.

Either we like the features in there, or at least some of them, and we should put those into async, or we don't, and we should drop them all.
Time to graduate from being an experiment :)

lib/src/stream_extensions.dart

nex3 added 2 commits February 7, 2023 17:49

Add StreamExtensions.buffer()

6fa57f1

Remove "new"

b489ceb

natebosch reviewed Feb 17, 2023

View reviewed changes

natebosch requested a review from lrhn February 17, 2023 02:34

lrhn approved these changes Feb 17, 2023

View reviewed changes

Code review

11f05d4

lrhn reviewed Feb 18, 2023

View reviewed changes

lib/src/stream_extensions.dart Show resolved Hide resolved

Code review

ba9f43b

lrhn approved these changes Mar 1, 2023

View reviewed changes

lib/src/stream_extensions.dart Outdated Show resolved Hide resolved

lib/src/stream_extensions.dart Outdated Show resolved Hide resolved

lib/src/stream_extensions.dart Show resolved Hide resolved

Code review

27c9694

nex3 requested a review from natebosch March 2, 2023 00:09

Update method name

f0b4670

natebosch approved these changes Mar 2, 2023

View reviewed changes

Format

cbeb54e

natebosch merged commit f454380 into dart-lang:master Mar 2, 2023

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add `StreamExtensions.buffer()` #235

Add `StreamExtensions.buffer()` #235

nex3 commented Feb 8, 2023

natebosch left a comment

natebosch Feb 17, 2023

natebosch Feb 17, 2023

nex3 Feb 17, 2023

lrhn Feb 17, 2023

jakemac53 Feb 17, 2023

lrhn Feb 18, 2023

natebosch Feb 23, 2023

lrhn Mar 1, 2023

natebosch Mar 2, 2023

nex3 Mar 2, 2023

natebosch Feb 17, 2023

nex3 Feb 17, 2023

natebosch Feb 17, 2023

natebosch commented Feb 17, 2023

nex3 commented Feb 17, 2023

lrhn left a comment

lrhn Feb 17, 2023

lrhn commented Feb 17, 2023

Add StreamExtensions.buffer() #235

Add StreamExtensions.buffer() #235

Conversation

nex3 commented Feb 8, 2023

natebosch left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

natebosch commented Feb 17, 2023

nex3 commented Feb 17, 2023

lrhn left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lrhn commented Feb 17, 2023

Add `StreamExtensions.buffer()` #235

Add `StreamExtensions.buffer()` #235