watch method

@override Stream<TaskStatus> watch(String taskId)

Watches updates for a task ID, emitting new statuses.

Implementation

@override
/// Watches updates for a task ID, emitting new statuses.
Stream<TaskStatus> watch(String taskId) {
  final controller = _watchers.putIfAbsent(
    taskId,
    () => StreamController<TaskStatus>.broadcast(
      onCancel: () {
        if (!(_watchers[taskId]?.hasListener ?? false)) {
          final controller = _watchers.remove(taskId);
          if (controller != null) {
            unawaited(controller.close());
          }
        }
      },
    ),
  );
  return controller.stream;
}