flutter_common/lib/upload_video/video_batch_upload_queue.dart

import 'dart:async';

import 'package:dio/dio.dart';
import 'package:flutter/widgets.dart';
import 'package:flutter_common/utils/toast_utils.dart';

import 'video_batch_upload_config.dart';
import 'video_batch_upload_service.dart';
import 'video_batch_upload_task.dart';

class VideoBatchUploadQueue {
  VideoBatchUploadQueue({
    required this.contextProvider,
    required this.uploadTask,
    required this.saveTask,
    required this.onChanged,
    this.onCompleted,
    this.maxAssets = VideoBatchUploadConfig.maxAssets,
    this.maxConcurrent = VideoBatchUploadConfig.maxConcurrent,
  });

  final BuildContext? Function() contextProvider;
  final VideoBatchUploadUploadTask uploadTask;
  final VideoBatchUploadSaveTask saveTask;
  final VoidCallback onChanged;
  final FutureOr<void> Function()? onCompleted;
  final int maxAssets;
  final int maxConcurrent;

  final List<VideoBatchUploadTask> tasks = <VideoBatchUploadTask>[];
  int _batchTotalCount = 0;
  int _completedRemovedCount = 0;
  bool isExpanded = false;
  bool _isRunning = false;
  bool _completionNotified = false;
  bool _isDisposed = false;

  bool get hasTasks => tasks.isNotEmpty;

  bool get hasActiveUploads => tasks.any((VideoBatchUploadTask task) {
        return task.isActive;
      });

  int get totalCount => _batchTotalCount > 0 ? _batchTotalCount : tasks.length;

  int get completedCount => tasks
      .where((VideoBatchUploadTask task) =>
          task.status == VideoBatchUploadStatus.completed)
      .length;

  int get failedCount => tasks
      .where((VideoBatchUploadTask task) =>
          task.status == VideoBatchUploadStatus.failed)
      .length;

  int get finishedCount =>
      _completedRemovedCount + completedCount + failedCount;

  Future<void> pickAndStart() async {
    if (hasActiveUploads) {
      isExpanded = true;
      _notifyChanged();
      ToastUtils.showToast(msg: '视频正在上传中');
      return;
    }

    final BuildContext? context = contextProvider();
    if (context == null || !context.mounted) {
      return;
    }

    final List<VideoBatchUploadTask> pickedTasks =
        await VideoBatchUploadPicker.pickVideos(
      context,
      maxAssets: maxAssets,
    );
    if (pickedTasks.isEmpty) {
      return;
    }

    tasks
      ..clear()
      ..addAll(pickedTasks);
    _batchTotalCount = pickedTasks.length;
    _completedRemovedCount = 0;
    isExpanded = true;
    _completionNotified = false;
    _notifyChanged();
    unawaited(_runQueue());
  }

  void toggleExpanded() {
    if (!hasTasks) {
      return;
    }
    isExpanded = !isExpanded;
    _notifyChanged();
  }

  void retryTask(VideoBatchUploadTask task) {
    if (!tasks.contains(task) || !task.canRetry) {
      return;
    }
    task
      ..status = VideoBatchUploadStatus.waiting
      ..progress = 0
      ..uploadUrl = ''
      ..errorMessage = '';
    _completionNotified = false;
    isExpanded = true;
    _notifyChanged();
    unawaited(_runQueue());
  }

  void removeTask(VideoBatchUploadTask task) {
    if (!tasks.contains(task)) {
      return;
    }
    task.cancelToken?.cancel('removed');
    tasks.remove(task);
    if (tasks.isEmpty) {
      isExpanded = false;
      _resetBatchProgress();
    }
    _notifyChanged();
  }

  void clear() {
    for (final VideoBatchUploadTask task in tasks) {
      task.cancelToken?.cancel('cleared');
    }
    tasks.clear();
    isExpanded = false;
    _isRunning = false;
    _resetBatchProgress();
    _notifyChanged();
  }

  void cancelAll() {
    for (final VideoBatchUploadTask task in tasks) {
      task.cancelToken?.cancel('cancelled');
    }
    for (final VideoBatchUploadTask task in tasks) {
      if (task.isActive) {
        task
          ..status = VideoBatchUploadStatus.failed
          ..errorMessage = '已取消';
      }
    }
    _isRunning = false;
    _notifyChanged();
  }

  void requestExit({required VoidCallback onExit}) {
    if (!hasActiveUploads) {
      onExit();
      return;
    }
    ToastUtils.showAlterDialog(
      titleText: '视频还在上传',
      contentText: '离开会中断本次上传，确定要离开吗？',
      confirmText: '离开',
      confirmCallback: () {
        cancelAll();
        onExit();
      },
    );
  }

  void dispose() {
    _isDisposed = true;
    for (final VideoBatchUploadTask task in tasks) {
      task.cancelToken?.cancel('disposed');
    }
  }

  void _resetBatchProgress() {
    _batchTotalCount = 0;
    _completedRemovedCount = 0;
    _completionNotified = false;
  }

  Future<void> _runQueue() async {
    if (_isRunning || _isDisposed) {
      return;
    }
    _isRunning = true;
    final int workerCount = maxConcurrent < 1 ? 1 : maxConcurrent;
    final List<Future<void>> workers = List<Future<void>>.generate(
      workerCount,
      (_) => _uploadWorker(),
    );
    await Future.wait(workers);
    _isRunning = false;
    await _notifyCompletedIfNeeded();
  }

  Future<void> _uploadWorker() async {
    while (true) {
      if (_isDisposed) {
        return;
      }
      final VideoBatchUploadTask? task = _nextWaitingTask();
      if (task == null) {
        return;
      }
      await _processTask(task);
    }
  }

  VideoBatchUploadTask? _nextWaitingTask() {
    for (final VideoBatchUploadTask task in tasks) {
      if (task.status == VideoBatchUploadStatus.waiting) {
        return task;
      }
    }
    return null;
  }

  Future<void> _processTask(VideoBatchUploadTask task) async {
    task
      ..status = VideoBatchUploadStatus.uploading
      ..progress = 0
      ..errorMessage = ''
      ..cancelToken = CancelToken();
    _notifyChanged();

    try {
      final String uploadUrl = await uploadTask(
        task,
        (double progress) {
          if (!tasks.contains(task)) {
            return;
          }
          task.progress = progress.clamp(0, 1);
          _notifyChanged();
        },
      );
      if (!tasks.contains(task)) {
        return;
      }
      if (uploadUrl.trim().isEmpty) {
        throw Exception('empty upload url');
      }
      task
        ..uploadUrl = uploadUrl
        ..progress = 1
        ..status = VideoBatchUploadStatus.uploaded;
      _notifyChanged();

      task.status = VideoBatchUploadStatus.saving;
      _notifyChanged();
      final bool saved = await saveTask(task);
      if (!tasks.contains(task)) {
        return;
      }
      if (saved) {
        task.status = VideoBatchUploadStatus.completed;
        _completedRemovedCount += 1;
        tasks.remove(task);
        if (tasks.isEmpty) {
          isExpanded = false;
        }
      } else {
        task
          ..status = VideoBatchUploadStatus.failed
          ..errorMessage = '入库失败';
      }
    } on DioException catch (error) {
      if (!tasks.contains(task)) {
        return;
      }
      task
        ..status = VideoBatchUploadStatus.failed
        ..errorMessage =
            CancelToken.isCancel(error) ? '已取消' : _networkErrorText(error);
    } catch (_) {
      if (!tasks.contains(task)) {
        return;
      }
      task
        ..status = VideoBatchUploadStatus.failed
        ..errorMessage = '上传失败';
    } finally {
      task.cancelToken = null;
      _notifyChanged();
    }
  }

  String _networkErrorText(DioException error) {
    switch (error.type) {
      case DioExceptionType.connectionTimeout:
      case DioExceptionType.sendTimeout:
      case DioExceptionType.receiveTimeout:
      case DioExceptionType.connectionError:
        return '网络异常';
      case DioExceptionType.cancel:
        return '已取消';
      case DioExceptionType.badResponse:
        return '上传失败';
      case DioExceptionType.badCertificate:
      case DioExceptionType.unknown:
        return '上传失败';
    }
  }

  Future<void> _notifyCompletedIfNeeded() async {
    if (_isDisposed ||
        _completionNotified ||
        totalCount == 0 ||
        hasActiveUploads) {
      return;
    }
    _completionNotified = true;
    await onCompleted?.call();
    if (failedCount > 0) {
      ToastUtils.showToast(
          msg: '已上传 $_completedRemovedCount 个，失败 $failedCount 个');
      return;
    }
    final int completed = _completedRemovedCount;
    ToastUtils.showToast(msg: '已上传 $completed 个视频');
    tasks.clear();
    isExpanded = false;
    _resetBatchProgress();
    _notifyChanged();
  }

  void _notifyChanged() {
    if (_isDisposed) {
      return;
    }
    onChanged();
  }
}