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, this.saveTask, this.batchSaveTask, required this.onChanged, this.assetType = VideoBatchUploadAssetType.video, this.onCompleted, this.maxAssets = VideoBatchUploadConfig.maxAssets, this.maxConcurrent = VideoBatchUploadConfig.maxConcurrent, }); final BuildContext? Function() contextProvider; final VideoBatchUploadUploadTask uploadTask; final VideoBatchUploadSaveTask? saveTask; final VideoBatchUploadBatchSaveTask? batchSaveTask; final VoidCallback onChanged; final VideoBatchUploadAssetType assetType; final FutureOr Function()? onCompleted; final int maxAssets; final int maxConcurrent; final List tasks = []; 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; bool get _usesBatchSave => batchSaveTask != null; Future pickAndStart() async { if (hasActiveUploads) { isExpanded = true; _notifyChanged(); ToastUtils.showToast(msg: '${assetType.label}正在上传中'); return; } final BuildContext? context = contextProvider(); if (context == null || !context.mounted) { return; } final List pickedTasks = await VideoBatchUploadPicker.pickMedia( context, maxAssets: maxAssets, assetType: assetType, ); 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: '${assetType.label}还在上传', 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 _runQueue() async { if (_isRunning || _isDisposed) { return; } _isRunning = true; final int workerCount = maxConcurrent < 1 ? 1 : maxConcurrent; final List> workers = List>.generate( workerCount, (_) => _uploadWorker(), ); await Future.wait(workers); _isRunning = false; await _saveUploadedBatchIfNeeded(); await _notifyCompletedIfNeeded(); } Future _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 _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(); if (_usesBatchSave) { return; } 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(); } } Future _saveUploadedBatchIfNeeded() async { if (!_usesBatchSave || _isDisposed) { return; } final List successfulUploadTasks = tasks .where((VideoBatchUploadTask task) => task.status == VideoBatchUploadStatus.uploaded) .toList(); if (successfulUploadTasks.isEmpty) { return; } for (final VideoBatchUploadTask task in successfulUploadTasks) { task.status = VideoBatchUploadStatus.saving; } _notifyChanged(); try { final bool saved = await batchSaveTask!(List.of( successfulUploadTasks, growable: false, )); if (_isDisposed) { return; } if (saved) { _completedRemovedCount += successfulUploadTasks.length; tasks.removeWhere(successfulUploadTasks.contains); if (tasks.isEmpty) { isExpanded = false; } return; } for (final VideoBatchUploadTask task in successfulUploadTasks) { if (tasks.contains(task)) { task ..status = VideoBatchUploadStatus.failed ..errorMessage = '入库失败'; } } } catch (_) { if (_isDisposed) { return; } for (final VideoBatchUploadTask task in successfulUploadTasks) { if (tasks.contains(task)) { task ..status = VideoBatchUploadStatus.failed ..errorMessage = '入库失败'; } } } finally { _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 _notifyCompletedIfNeeded() async { if (_isDisposed || _completionNotified || totalCount == 0 || hasActiveUploads) { return; } _completionNotified = true; await onCompleted?.call(); if (failedCount > 0) { ToastUtils.showToast( msg: '已上传 $_completedRemovedCount ${assetType.unit},失败 $failedCount ${assetType.unit}'); return; } final int completed = _completedRemovedCount; ToastUtils.showToast( msg: '已上传 $completed ${assetType.unit}${assetType.label}'); tasks.clear(); isExpanded = false; _resetBatchProgress(); _notifyChanged(); } void _notifyChanged() { if (_isDisposed) { return; } onChanged(); } }