feat: enable concurrency 2 for sending

app/test/unit/util/task_runner_test.dart

@@ -1,37 +0,0 @@
-import 'package:localsend_isolates/util/sleep.dart';
-import 'package:localsend_isolates/util/task_runner.dart';
-import 'package:test/test.dart';
-
-void main() {
-  group('TaskRunner', () {
-    test('should run all tasks in parallel', () async {
-      final results = TaskRunner<String?>(
-        concurrency: 10,
-        initialTasks: [
-          for (final data in [
-            [10, null],
-            [30, 'a'],
-            [20, 'b'],
-            [40, 'c'],
-          ])
-            () async {
-              final delay = data[0] as int;
-              final result = data[1] as String?;
-              await sleepAsync(delay);
-              return result;
-            },
-        ],
-      ).stream;
-
-      String? finalResult;
-
-      await results.forEach((result) {
-        if (finalResult == null && result != null) {
-          finalResult = result;
-        }
-      });
-
-      expect(finalResult, 'b');
-    });
-  });
-}

packages/localsend_isolates/lib/src/isolate/child/upload_isolate.dart

@@ -7,9 +7,13 @@ import 'package:localsend_isolates/src/isolate/dto/send_to_isolate_data.dart';
 import 'package:localsend_isolates/src/task/upload/http_upload.dart';
 import 'package:localsend_isolates/util/android_channel.dart';
 import 'package:localsend_isolates/util/rust.dart';
+import 'package:pool/pool.dart';
 import 'package:refena_flutter/refena_flutter.dart';
 import 'package:typed_isolates/typed_isolates.dart';
 
+/// How many files of a [HttpUploadFilesTask] are uploaded in parallel.
+const _concurrency = 2;
+
 sealed class BaseHttpUploadTask {}
 
 class HttpUploadFile {
@@ -31,8 +35,8 @@ class HttpUploadFile {
 /// Uploads a list of files as one isolate task.
 ///
 /// This task is intended to replace the file scheduling loop in the parent
-/// isolate. Files are uploaded sequentially and progress is reported across
-/// the complete list.
+/// isolate. Up to [_concurrency] files are uploaded in parallel and progress
+/// is reported across the complete list.
 class HttpUploadFilesTask implements BaseHttpUploadTask {
   final String? remoteSessionId;
   final List<HttpUploadFile> files;
@@ -126,7 +130,12 @@ Future<void> setupHttpUploadIsolate(
       final cancelToken = createCancellationToken();
       ref.read(_cancelTokenProvider).putIfAbsent(task.id, () => cancelToken);
       try {
-        for (final file in uploadTask.files) {
+        await Pool(_concurrency).forEach<HttpUploadFile, void>(uploadTask.files, (file) async {
+          if (!ref.read(_cancelTokenProvider).containsKey(task.id)) {
+            // the task was canceled, do not upload the remaining files
+            return;
+          }
+
           sendToMain(
             IsolateTaskStreamResult.event(
               id: task.id,
@@ -181,12 +190,7 @@ Future<void> setupHttpUploadIsolate(
               ),
             );
           }
-
-          if (!ref.read(_cancelTokenProvider).containsKey(task.id)) {
-            // the task was canceled, do not upload the remaining files
-            break;
-          }
-        }
+        }).drain<void>();
 
         sendToMain(
           IsolateTaskStreamResult.done(

packages/localsend_isolates/lib/src/task/discovery/http_scan_discovery.dart

@@ -1,18 +1,25 @@
 import 'package:localsend_isolates/model/device.dart';
 import 'package:localsend_isolates/src/task/discovery/http_target_discovery.dart';
-import 'package:localsend_isolates/util/task_runner.dart';
 import 'package:logging/logging.dart';
+import 'package:pool/pool.dart';
 import 'package:refena_flutter/refena_flutter.dart';
 
 final _logger = Logger('HttpScanDiscovery');
 
+const _concurrency = 50;
+
 final httpScanDiscoveryProvider = ViewProvider((ref) {
   return HttpScanDiscoveryService(
     targetedDiscoveryService: ref.accessor(httpTargetDiscoveryProvider),
   );
 });
 
-Map<String, TaskRunner> _runners = {};
+class _CancelToken {
+  bool cancelled = false;
+}
+
+/// The token of the currently running scan per network interface.
+Map<String, _CancelToken> _cancelTokens = {};
 
 class HttpScanDiscoveryService {
   final StateAccessor<HttpTargetDiscoveryService> _targetedDiscoveryService;
@@ -23,32 +30,18 @@ class HttpScanDiscoveryService {
 
   Stream<Device> getStream({required String networkInterface, required int port, required bool https}) {
     final ipList = List.generate(256, (i) => '${networkInterface.split('.').take(3).join('.')}.$i').where((ip) => ip != networkInterface).toList();
-    _runners[networkInterface]?.stop();
-    _runners[networkInterface] = TaskRunner<Device?>(
-      initialTasks: List.generate(
-        ipList.length,
-        (index) =>
-            () async => _doRequest(ipList[index], port, https),
-      ),
-      concurrency: 50,
-    );
 
-    return _runners[networkInterface]!.stream.where((device) => device != null).cast<Device>();
+    // Let the previous scan of this interface skip its remaining requests, so its stream ends.
+    _cancelTokens[networkInterface]?.cancelled = true;
+    final token = _cancelTokens[networkInterface] = _CancelToken();
+
+    final stream = Pool(_concurrency).forEach<String, Device?>(ipList, (ip) async => token.cancelled ? null : _doRequest(ip, port, https));
+    return stream.where((device) => device != null).cast<Device>();
   }
 
   Stream<Device> getFavoriteStream({required List<(String, int)> devices, required bool https}) {
-    final runner = TaskRunner<Device?>(
-      initialTasks: List.generate(
-        devices.length,
-        (index) => () async {
-          final device = devices[index];
-          return _doRequest(device.$1, device.$2, https);
-        },
-      ),
-      concurrency: 50,
-    );
-
-    return runner.stream.where((device) => device != null).cast<Device>();
+    final stream = Pool(_concurrency).forEach<(String, int), Device?>(devices, (device) => _doRequest(device.$1, device.$2, https));
+    return stream.where((device) => device != null).cast<Device>();
   }
 
   Future<Device?> _doRequest(String currentIp, int port, bool https) async {

packages/localsend_isolates/lib/util/task_runner.dart

@@ -1,69 +0,0 @@
-import 'dart:async';
-import 'dart:collection';
-
-typedef FutureFunction<T> = Future<T> Function();
-
-class TaskRunner<T> {
-  final StreamController<T> _streamController = StreamController();
-  final Queue<FutureFunction<T>> _queue;
-
-  final int concurrency;
-  int _runnerCount = 0;
-  bool _stopped = false;
-
-  /// If [true], then the stream will be closed as soon as every task has been finished.
-  /// By default, it is [false] when [initialTasks] is provided with a non-empty list.
-  final bool _stayAlive;
-
-  final void Function()? onFinish;
-
-  TaskRunner({
-    required this.concurrency,
-    List<FutureFunction<T>>? initialTasks,
-    bool? stayAlive,
-    this.onFinish,
-  }) : _queue = Queue()..addAll(initialTasks ?? []),
-       _stayAlive = stayAlive ?? initialTasks == null || initialTasks.isEmpty {
-    _fireRunners();
-  }
-
-  void addAll(Iterable<FutureFunction<T>> iterable) {
-    _queue.addAll(iterable);
-    _fireRunners();
-  }
-
-  void stop() {
-    _stopped = true;
-  }
-
-  Stream<T> get stream => _streamController.stream;
-
-  /// Starts multiple runners until [concurrency].
-  void _fireRunners() {
-    while (_queue.isNotEmpty && _runnerCount < concurrency && !_streamController.isClosed) {
-      _runnerCount++;
-
-      unawaited(
-        _runner(
-          onFinish: () {
-            _runnerCount--;
-            if (_stopped || (_runnerCount == 0 && !_stayAlive)) {
-              // ignore: discarded_futures
-              _streamController.close();
-              onFinish?.call();
-            }
-          },
-        ),
-      );
-    }
-  }
-
-  /// Processes the queue one by one.
-  Future<void> _runner({required void Function() onFinish}) async {
-    while (_queue.isNotEmpty) {
-      final task = _queue.removeFirst();
-      _streamController.add(await task());
-    }
-    onFinish();
-  }
-}

packages/localsend_isolates/pubspec.lock

@@ -327,10 +327,10 @@ packages:
     dependency: transitive
     description:
       name: meta
-      sha256: "23f08335362185a5ea2ad3a4e597f1375e78bce8a040df5c600c8d3552ef2394"
+      sha256: "1741988757a65eb6b36abe716829688cf01910bbf91c34354ff7ec1c3de2b349"
       url: "https://pub.dev"
     source: hosted
-    version: "1.17.0"
+    version: "1.18.0"
   mime:
     dependency: "direct main"
     description:
@@ -396,7 +396,7 @@ packages:
     source: hosted
     version: "2.1.8"
   pool:
-    dependency: transitive
+    dependency: "direct main"
     description:
       name: pool
       sha256: "978783255c543aa3586a1b3c21f6e9d720eb315376a915872c61ef8b5c20177d"

packages/localsend_isolates/pubspec.yaml

@@ -21,6 +21,7 @@ dependencies:
   logging: 1.3.0
   mime: 2.0.0
   path: 1.9.1
+  pool: 1.5.2
   refena_flutter: 3.2.1
   rust_lib_localsend_app:
     path: rust_builder