Possible to insert values into stream asynchronously in dart?-CodePudding

I have a function which downloads the RSS Feed and parses it. After that, I iterate through each team and do some processing on it. For now, this processing is asynchronous and whenever it completes I yield it but that means each item will be processed linearly. Is there a way to process all the items once and yield them as they are processed?

High level code:

Stream<String> processRSSFeedItems() async* {
    RSSFeed rssFeed = RSSFeed("some URL");
    for(int itemIndex = 0; itemIndex < rssFeed.items.lengthl itemIndex  ){
         String processedItem = await processRssItem(rssFeed.items[itemIndex]);
         yield processedItem;
    }
}

void refreshRssFeed(){
    var stream = processRSSFeedItems();
    stream.listen((item) => setState(() => something(item));
}

Is there any way to call processRssItem() for all items without waiting for the previous process to complete and yield them as they arrive?

CodePudding user response：

You can do

List<Future<String>> futures = <Future<String>>[];
for(int itemIndex = 0; itemIndex < rssFeed.items.length; itemIndex  ){
     futures.add(processRssItem(rssFeed.items[itemIndex]));
}

// Waits for multiple futures to complete and collects their results.
final result = await Future.wait<String>(futures);

for(var i = 0 ; i < result.length; i  ) {
    yield result[i];
}

CodePudding user response：

You could add the elements to the sink of a StreamController as they arrive, and return the stream of the StreamController.

import 'dart:async';
...

Stream<String> processRSSFeedItems() {
  StreamController<String> controller = StreamController<String>();
  RSSFeed rssFeed = RSSFeed("some URL");
  for (int itemIndex = 0; itemIndex < rssFeed.items.length; itemIndex  ) {
    processRssItem(rssFeed.items[itemIndex]).then(controller.sink.add);
  }
  return controller.stream.take(rssFeed.items.length);
}

CodePudding user response：

From your question:

"For now, this processing is asynchronous and whenever it completes I yield it but that means each item will be processed linearly. Is there a way to process all the items once and yield them as they are processed?"

"yield" in this context is equivalent to "return". So Stream will not fit your use-case because Streams are designed to be listened for values iteratively, not for "all the items once". For "all the items once" you should use Future and return the value, not yield. Updating Rahul's answer

    Future getAllWhenProcessed(){
      List<Future<String>> futures = <Future<String>>[];
      for(int itemIndex = 0; itemIndex < rssFeed.items.length;   itemIndex  ){
         futures.add(processRssItem(rssFeed.items[itemIndex]));
      }

      // Waits for multiple futures to complete and collects their results.
      List results = [];
      for(var future in futures){
         results.add(await future);
      }
      return results;
    }

Then just

getAllWhenProcessed().then((value){ updateUi(); });