Pause an async function until a condition is met in dart-CodePudding

I have a widget that performs a series of expensive computations (which are carried out on a compute core). The computations are queued in a list in the widget and the computation of each is waited for. The computations are queued in the build function so since before being sent to the compute core require some computations to be performed on the main isolate, I thought not to handle them directly in the build function but to just add them to the queue at build time.

This is the function I currently have to exhaust the queue:

void emptyQueue() async {

  while(queue.isNotEmpty) {
    /* Perform some computations here, which prevents me from just sending the data to the compute core directly (they need data that is only in the main isolate and it's too big to send to the compute core) */

    /* Send the data to the worker isolate */
    await Future.delayed(Duration(milliseconds: 1000));

    /* Remove the first element from the queue. */
    queue.removeFirst();
  }
}

This function works well when the function is called as the queue is fixed and no more elements are added to it afterward as it just loops until the queue is empty. My problem is that I can add requests to the queue at any point in time, even when the queue is empty, and by then the function would be completely done and thus the queue would not be exhausted anymore.

I thought something like this might work but so far I wasn't able to concretely implement it.

void emptyQueue() async {
    while (true) {
      if (queue.isNotEmpty()) {
        /* Perform some computations here (gather data to send to the compute core) */

        /* Send the data to the worker isolate */
        await Future.delayed(Duration(milliseconds: 1000));

        /* Remove the first element from the queue. */
        queue.removeFirst();
      }

      await /* a signal of some sort */;
    }
  }

In this way, I could start the emptyQueue in the initState as it would be paused until something is added to the queue. Moreover, the function which would be called at build time would be just queue.add(item), without any of the main isolate-side computations.

Is there any way I could pause the function at the end (at the last await)?

CodePudding user response：

Streams are the best fit for such scenarios. Consider the following..

import 'dart:async';

// creating a stream
StreamController<String> streamController = StreamController();

void main(){
  
  // stream listener
  streamController.stream.listen((value){
    print(value); // print message to terminal
  });
  
  // adding object to stream
  streamController.add("hello");
  streamController.add("bye");
  streamController.add("seeya");
}

Here, whenever you add an object (in this case a string) into the stream, the listener is called which executes whatever you have defined inside it. In my case, the code will print hello bye and seeya to the terminal.

Similarly, you can use stream to fire up some piece of code whenever you add something to the stream.

CodePudding user response：

I was able to pause the emptyQueue by awaiting for a Completer.

void emptyQueue() async {
    while (true) {
      if (queue.isNotEmpty()) {
        /* Perform some computations here (gather data to send to the compute core) */

        /* Send the data to the worker isolate */
        await Future.delayed(Duration(milliseconds: 1000));

        /* Remove the first element from the queue. */
        queue.removeFirst();
      }

      /* Waiting for queue items */
      if (!_queueFreezerCompleter.isCompleted) _queueFreezerCompleter.complete();
      _queueFreezerCompleter = Completer();

      await _queueFreezerCompleter.future;
    }
  }

The completer is just an attribute of the State class.

Completer _queueFreezerCompleter = Completer();

This Completer is completed (only if it was not completed already) when the new item is added to the queue, meaning that the external loop while(true) will restart and the queue will be emptied again.

void onAddQueue(){
   queue.enqueue(value);
   if (!_queueFreezerCompleter.isCompleted) _queueFreezerCompleter.complete();
}

With this solution is also possible to remove items from the queue (if they haven't been processed already)