FirstWhere doesn't complete even though test returns true

[LeoBound]

I am trying to write a flutter app that makes fairly extensive use of Streams and Futures as it retrieves data over bluetooth.

I have a Stream that I create with an async * function by mutating another stream. I want to get the first element of this stream that matches some critera so I use firstWhere which should complete a Future when the condition is met.

Although the condition function returns true, the Future doesn't complete until the next element is processed by the stream.

I've written this example to demonstrate what I mean:
https://dartpad.dev/12ae241a54f567cea44b53fc7e1c7412

This doesn't occur when using the underlying stream directly however.
(replace line 4 with packetStream.where(testPacket).listen((packet) => print("$packet Matches!"));)

Output of "dart --version":

Dart SDK version: 2.14.4 (stable) (Wed Oct 13 11:11:32 2021 +0200) on "windows_x64"

Although as you can see this also occurs on DartPad.

Is there a way around this as the behaviour seems counterintuitive?

[mraleph]

/cc @lrhn

[lrhn]

Sounds like a variant of the "async* functions being implemented incorrectly"-issue (#34775).

The firstWhere cancels the stream subscription when it sees the matching event. Then it awaits the future returned by cancel. That's the problem here: The implementation of async*'s yield moves on too quickly, it doesn't wait for the event to be delivered, and then check for whether the code responding to the event calls cancel.

So, the async* function moves on, and is unable to cancel until it reaches the next yield.

We should just fix it ... but it's hard-wired fairly deeply into the VM's async* implementation, so it needs someone familiar with that to make the change.

[LeoBound]

Is there a way around this right now as I'm unable to use Flutter's FutureBuilder as a lot of the time my stream may not receive a 2nd packet for a while after the firstWhere packet.

EDIT:
As a workaround it looks like you can do something like this for now:

import 'dart:async';

void main() {
  
  final intStream = Stream.periodic(Duration(seconds: 1), (i) => i);
  final packetStream = s.bind(intStream);
  packetStream.firstWhere(testPacket).then((packet) => print("$packet Matches!"));
}

final s = StreamTransformer<int, String>.fromHandlers(handleData: (int data, EventSink<String> sink) {
    print("Processing $data");
    sink.add("Packet $data");
  });

bool testPacket(String packet) {
  final result = packet == "Packet 4";
  print(" $packet passed? : $result");
  return result;
}