timeout on a stream firstWhere call doesn't work

[VKBobyr]

I'm trying to listen to a stream until I find a certain value or timeout after 1 second.
Here's a simple example:

 import 'dart:io';
 
 Stream<int> data() async* {
   int count = 1;

   // counts from 1 to infinity in 1 second intervals
   while (true) {
     sleep(Duration(seconds: 1));
     yield count;
     print('Stream: $count');
     count++;
   }
 }
 
 
 void main() async {
     // waits for the stream to emit a 5 and returns it or times out after 1 second and returns null
     final found = await data().firstWhere((e)=>e == 5)
                           .timeout(Duration(seconds: 1), onTimeout: ()=>null);
     print('found $found');
 }

Expected behavior
Listens to the stream for 1 second and times out with a null value (because the timeout is set to 1 second and it takes 5 seconds to get to 5).
Expected output: found null

Actual Behavior
Waits for 5 seconds for the stream to emit 5, despite the timeout() setting.
Actual output: found 5

[julemand101]

Your problem is that your data() method is really not async but instead operates as a normal sync method which sleeps the whole isolate for each iteration. Since all your code are running inside a single thread, which you are sleeping, then Dart cannot execute other logic in the same isolate like e.g. inner logic behind timeout.

This behavior was introduced with Dart 2.0.0 and mentioned in the changelog:

(Breaking) Functions marked async now run synchronously until the first await statement. Previously, they would return to the event loop once at the top of the function body before any code runs (#30345).

There are also a warning in the API for the sleep method:

Use this with care, as no asynchronous operations can be processed in a isolate while it is blocked in a sleep call.

https://api.dart.dev/stable/2.9.3/dart-io/sleep.html

If you rewrite you code so it does not make use of sleep but instead uses Future.delayed which allows Dart to still process other tasks on the task queue and makes your data method truly async:

Stream<int> data() async* {
  int count = 1;

  // counts from 1 to infinity in 1 second intervals
  while (true) {
    await Future<void>.delayed(const Duration(seconds: 1));
    yield count;
    print('Stream: $count');
    count++;
  }
}

void main() async {
  // waits for the stream to emit a 5 and returns it or times out after 1 second and returns null
  final found = await data()
      .firstWhere((e) => e == 5)
      .timeout(const Duration(seconds: 1), onTimeout: () => null);
  print('found $found');
}

You will then get the following output:

found null
Stream: 1
Stream: 2
Stream: 3
Stream: 4
Stream: 5

Note that the data method is not stopped since firstWhere will still be executed until it is done. But the result are trowed away.

[lrhn]

Working as intended.

The Future returned by firstWhere has no way to cancel the underlying operation. When you do a timeout on that future, you create a new future which completes with either the timeout or the value of the original future, but even with the timeout, you won't cancel the underlying computation, and the original future will eventually complete.

(No opinion on the use of sleep in this, but the futures and streams are working as they should).