flatMapBatchesSingle.dart

import 'package:rxdart_ext/rxdart_ext.dart';
import 'package:tuple/tuple.dart';

Future<List<int>> getList() async {
  print('getList: start');
  await delay(500);
  print('getList: end');

  return List.generate(10, identity);
}

Future<Tuple2<int, String>> getPerItem(int item) async {
  print('getPerItem $item: start');
  await delay(1000);
  print('getPerItem $item: end');

  return Tuple2(item, 'Result of $item');
}

Future<void> work() async {
  final watch = Stopwatch()..start();

  final batchSize = 4;
  final maxRetries = 3;
  final List<Tuple2<int, String>> result = await getList()
      .asSingle()
      .exhaustMap((items) => Stream.fromIterable(items))
      .flatMapBatchesSingle(
        (item) => Single.retry(() => getPerItem(item).asSingle(), maxRetries),
        batchSize,
      )
      .first;

  print(result);
  print('Total elapsed: ${watch.elapsedMilliseconds}ms');
}

void main() async {
  await work();
}

// [[0, Result of 0], [1, Result of 1], [2, Result of 2], [3, Result of 3], ... [8, Result of 8], [9, Result of 9]]
// Total elapsed: 3566ms

pubspec.yaml

name: dart_playground
description: A sample command-line application.
version: 1.0.0
# homepage: https://www.example.com

environment:
  sdk: '>=2.18.2 <3.0.0'

dependencies:
  rxdart_ext: ^0.2.7
  tuple: ^2.0.1

dev_dependencies:
  lints: ^2.0.0
  test: ^1.16.0