Possible regression with async* generator with dart 2.18.0 changes. Not yielding all events with yield*.

[Pacane]

Environment details:

Before we were using Dart 2.17.6 with Flutter 3.0.5
Now we're using Dart 2.18.2 with Flutter 3.3.4.

Got this issue running tests using the dartvm. I haven't ran the test cases against web at that time.

Problem

Before dart 2.18.0 async* changes We were using this implementation to create a stream seeded with an initial value:

Stream<T> createSeededStream<T>(T initialValue, Stream<T> stream) async* {
  yield initialValue;
  yield* stream;
}

Which used to work fine for us. I believe we could get this to work fine for both Streams and BroadcastStreams. However since the 2.18.0 change, this implementation didn't work anymore. Our test cases started failing, dropping sometimes the initial value, sometimes the first values of the yielded stream.

We found that using this implementation fixes the issue, but I'm not quite sure why it's not working all the time anymore.

Stream<T> createSeededStream<T>(T initialValue, Stream<T> stream) =>
    Stream.value(initialValue).mergeWith([stream]);

I believe the issue might lie in the fact that stream could be a broadcast stream, and that sometimes in a test environment the initial elements emitted in the stream are emitted too fast perhaps?

I'm not sure how/where to go from here to help you figure out if there is a regression for real, but these were our observations.

cc: @kevmoo

[kevmoo]

In what environments are you hitting this? web? native?

CC @lrhn – who likely has some context here!

[lrhn]

It shouldn't be possible to drop values. If there is a yield in the async* method, then it will definitely yield a value.

What might happen here is that the initial yield now waits until the value has been delivered before it continues, where before it would fire-and-forget the event and immediately start listening on stream afterwards. (That behavior was a bug).

If stream is a broadcast stream, then any events emitted by stream during that delay will not be part of the async* stream, because they happen before reaching the yield* stream;.

There may also be other delays involved. Calling listen on an async* stream doesn't start the body running immediately, so even doing createSeededStream(seed, stream).forEach(print) may drop events emitted by stream between the call to forEach and the body starting to run.

A possible workaround might be:

Stream<T> createSeededStream<T>(T initialValue, Stream<T> stream) async* {
  var c = StreamController<T>(sync: true);
  stream.pipe(c).ignore(); // Buffer events happening until yield*
  yield initialValue;
  yield* c.stream; // Emit events of `stream`, buffered ones first.
}

[Pacane]

If stream is a broadcast stream, then any events emitted by stream during that delay will not be part of the async* stream, because they happen before reaching the yield* stream;.

From my initial understanding/guess from the issue, that's what I thought happened. In our case, the initialization part of the stream remained the same. The behaviour just changed over night with 2.18.0. Which, could be the right behaviour now and used to be a bug as you said.

Regardless, we have a fix now, I was mostly wondering if it was a caveat to have in mind when using async* with broadcast streams.

[lrhn]

Everything about broadcasts stream is a caveat. I'd go as far as saying: Don't use a broadcast stream if you can't accept losing some events.

Even the delay between calling listen on a string returned by an async* method, and the body actually executing, is enough to lose events. (That's one of the reasons none of the methods on Stream are written as async* methods. They really should start running synchronously, like async does.)

If you want to emit events to multiple listeners, but not necessarily the same events at the same time, I'd try using Stream.multi for it, instead of a broadcast stream.
It's more work to write, because you don't get that from async* (I wish you did, but you don't so far).

[mraleph]

I am going ahead and close this issue given that it seems to have been answered.