Refine TaskQueue pauseAll to support granular pausing (#564)

* Refine TaskQueue pauseAll/resumeAll to support specific tasks and groups (#563)

- Updated TaskQueue interface to accept optional `tasks` and `group` parameters in `pauseAll` and `resumeAll`.
- Implemented granular pause logic in `MemoryTaskQueue` using a `_pausedTaskIds` set to track specific paused tasks.
- Updated `FileDownloader` to pass these parameters to the task queues.
- Ensures backward compatibility: calling `pauseAll()` without arguments still performs a global pause.
- Added unit tests in `test/task_queue_test.dart` to verify granular pause behavior.

Author: google-labs-jules[bot]

lib/src/file_downloader.dart

@@ -687,7 +687,7 @@ interface class FileDownloader {
   Future<List<DownloadTask>> pauseAll(
       {Iterable<DownloadTask>? tasks, String? group}) {
     for (final taskQueue in _downloader.taskQueues) {
-      taskQueue.pauseAll();
+      taskQueue.pauseAll(tasks: tasks, group: group);
     }
     return _downloader.pauseAll(tasks: tasks, group: group);
   }
@@ -729,7 +729,7 @@ interface class FileDownloader {
       await Future.delayed(interval);
     }
     for (final taskQueue in _downloader.taskQueues) {
-      taskQueue.resumeAll();
+      taskQueue.resumeAll(tasks: tasks, group: group);
     }
     return results;
   }

lib/src/queue/task_queue.dart

@@ -12,17 +12,24 @@ abstract interface class TaskQueue {
   /// Signals that [task] has finished
   void taskFinished(Task task);
 
-  /// Pauses all task processing in the queue.
+  /// Pauses task processing in the queue.
+  ///
+  /// If [tasks] or [group] are provided, pauses only those tasks.
+  /// If both are null, pauses all tasks.
   ///
   /// Default implementation is a no-op to ensure backwards compatibility
   /// with subclasses that don't override this method
-  Future<void> pauseAll() async {}
+  Future<void> pauseAll({Iterable<DownloadTask>? tasks, String? group}) async {}
 
-  /// Resumes all task processing in the queue
+  /// Resumes task processing in the queue
+  ///
+  /// If [tasks] or [group] are provided, resumes only those tasks.
+  /// If both are null, resumes all tasks.
   ///
   /// Default implementation is a no-op to ensure backwards compatibility
   /// with subclasses that don't override this method
-  Future<void> resumeAll() async {}
+  Future<void> resumeAll(
+      {Iterable<DownloadTask>? tasks, String? group}) async {}
 }
 
 /// TaskQueue that holds all information in memory
@@ -56,19 +63,49 @@ class MemoryTaskQueue implements TaskQueue {
   final _enqueueErrorsStreamController = StreamController<Task>();
 
   var _paused = false;
+  final _pausedTaskIds = <String>{};
 
   MemoryTaskQueue() {
     _readyForEnqueue.complete();
   }
 
   @override
-  Future<void> pauseAll() async {
-    _paused = true;
+  Future<void> pauseAll({Iterable<DownloadTask>? tasks, String? group}) async {
+    if (tasks == null && group == null) {
+      _paused = true;
+    } else {
+      // pause specific tasks/groups
+      if (group != null) {
+        final tasksToPause = waiting.unorderedElements
+            .where((task) => task.group == group && task is DownloadTask);
+        _pausedTaskIds.addAll(tasksToPause.map((e) => e.taskId));
+      }
+      if (tasks != null) {
+        _pausedTaskIds.addAll(tasks.map((e) => e.taskId));
+      }
+    }
   }
 
   @override
-  Future<void> resumeAll() async {
-    _paused = false;
+  Future<void> resumeAll({Iterable<DownloadTask>? tasks, String? group}) async {
+    if (tasks == null && group == null) {
+      _paused = false;
+      _pausedTaskIds.clear();
+    } else {
+      // resume specific tasks/groups
+      if (group != null) {
+        final tasksToResume = waiting.unorderedElements
+            .where((task) => task.group == group && task is DownloadTask);
+        for (final task in tasksToResume) {
+          _pausedTaskIds.remove(task.taskId);
+        }
+      }
+      if (tasks != null) {
+        for (final task in tasks) {
+          _pausedTaskIds.remove(task.taskId);
+        }
+      }
+    }
     advanceQueue();
   }
 
@@ -171,6 +208,10 @@ class MemoryTaskQueue implements TaskQueue {
     final tasksThatHaveToWait = <Task>[];
     while (waiting.isNotEmpty) {
       var task = waiting.removeFirst();
+      if (_pausedTaskIds.contains(task.taskId)) {
+        tasksThatHaveToWait.add(task);
+        continue;
+      }
       if (numActiveWithHostname(task.hostName) < maxConcurrentByHost &&
           numActiveWithGroup(task.group) < maxConcurrentByGroup) {
         waiting.addAll(tasksThatHaveToWait); // put back in queue

test/task_queue_test.dart

@@ -31,15 +31,15 @@ const workingUrl = 'https://google.com';
 
 void main() {
   WidgetsFlutterBinding.ensureInitialized();
-  final tq = TestTaskQueue();
+  late TestTaskQueue tq;
 
   var task = DownloadTask(url: 'testUrl');
 
   setUp(() {
+    tq = TestTaskQueue();
     tq.probFailure = 0;
     tq.maxConcurrent = 10000000;
     tq.minInterval = const Duration(milliseconds: 550);
-    tq.reset();
   });
 
   group('Add to queue', () {
@@ -215,4 +215,97 @@ void main() {
       expect(errorCount, greaterThan(3));
     });
   });
+
+  group('Pause and Resume', () {
+    test('pauseAll global', () async {
+      tq.maxConcurrent = 2;
+      for (var n = 0; n < 10; n++) {
+        task = DownloadTask(taskId: '$n', url: 'testUrl');
+        tq.add(task);
+      }
+      expect(tq.waiting.length, equals(9));
+
+      // Initial processing
+      await Future.delayed(const Duration(seconds: 1));
+      // Should have dequeued 2 tasks
+      expect(tq.waiting.length, lessThan(10));
+      expect(tq.numActive, equals(2));
+
+      await tq.pauseAll();
+
+      // Wait a bit - queue should not advance
+      final waitingBefore = tq.waiting.length;
+      await Future.delayed(const Duration(seconds: 6)); // wait for active tasks to finish
+      // Active tasks finish, but no new tasks should be dequeued
+      expect(tq.waiting.length, equals(waitingBefore));
+      expect(tq.numActive, equals(0));
+
+      await tq.resumeAll();
+      await Future.delayed(const Duration(seconds: 1));
+      expect(tq.numActive, equals(2));
+    });
+
+    test('pauseAll by group', () async {
+      tq.maxConcurrent = 4;
+      for (var n = 0; n < 5; n++) {
+        task = DownloadTask(taskId: 'A-$n', url: 'testUrl', group: 'A');
+        tq.add(task);
+      }
+      for (var n = 0; n < 5; n++) {
+        task = DownloadTask(taskId: 'B-$n', url: 'testUrl', group: 'B');
+        tq.add(task);
+      }
+
+      // Pause group A immediately
+      await tq.pauseAll(group: 'A');
+
+      // Wait for queue to process
+      await Future.delayed(const Duration(seconds: 2));
+
+      // Tasks from group A should be waiting, group B should be active
+      expect(tq.numActiveWithGroup('B'), greaterThan(0));
+      expect(tq.numActiveWithGroup('A'), lessThan(2)); // A-0 may be active
+
+      // Wait for B to finish
+      await Future.delayed(const Duration(seconds: 10));
+      expect(tq.numActiveWithGroup('B'), equals(0));
+
+      // A should still be waiting
+      expect(tq.numWaitingWithGroup('A'), equals(4));
+
+      await tq.resumeAll(group: 'A');
+      await Future.delayed(const Duration(seconds: 1));
+      expect(tq.numActiveWithGroup('A'), greaterThan(0));
+    });
+
+    test('pauseAll by specific tasks', () async {
+      tq.maxConcurrent = 5;
+      final tasksA = <DownloadTask>[];
+      final tasksB = <DownloadTask>[];
+
+      for (var n = 0; n < 5; n++) {
+        var t = DownloadTask(taskId: 'A-$n', url: 'testUrl', group: 'A');
+        tasksA.add(t);
+        tq.add(t);
+      }
+      for (var n = 0; n < 5; n++) {
+        var t = DownloadTask(taskId: 'B-$n', url: 'testUrl', group: 'B');
+        tasksB.add(t);
+        tq.add(t);
+      }
+
+      // Pause specific tasks from group A
+      await tq.pauseAll(tasks: tasksA);
+
+      await Future.delayed(const Duration(seconds: 2));
+
+      // Group B should be running, Group A should be paused
+      expect(tq.numActiveWithGroup('B'), greaterThan(0));
+      expect(tq.numActiveWithGroup('A'), lessThan(2)); // A-0 may be active
+
+      await tq.resumeAll(tasks: tasksA);
+      await Future.delayed(const Duration(seconds: 2));
+      expect(tq.numActiveWithGroup('A'), greaterThan(0));
+    });
+  });
 }