diff --git a/lib/app/layouts/conversation_view/widgets/message/attachment/attachment_holder.dart b/lib/app/layouts/conversation_view/widgets/message/attachment/attachment_holder.dart index 3c89499897..aa1dd95652 100644 --- a/lib/app/layouts/conversation_view/widgets/message/attachment/attachment_holder.dart +++ b/lib/app/layouts/conversation_view/widgets/message/attachment/attachment_holder.dart @@ -104,15 +104,28 @@ class _AttachmentHolderState extends CustomState(tag: _content.attachment.guid); + final downloader = attachmentDownloader.startDownload( + _content.attachment, + onComplete: onComplete, + prioritized: true, + ); setState(() { - content = attachmentDownloader.startDownload(_content.attachment, onComplete: onComplete); + content = downloader; }); } }, diff --git a/lib/services/network/attachment_download_queue.dart b/lib/services/network/attachment_download_queue.dart new file mode 100644 index 0000000000..43f621f734 --- /dev/null +++ b/lib/services/network/attachment_download_queue.dart @@ -0,0 +1,49 @@ +class AttachmentDownloadQueue { + final Map> _queues = >{}; + + Iterable get all => _queues.values.expand((items) => items); + + List forChat(String chatGuid) => + List.unmodifiable(_queues[chatGuid] ?? []); + + void add(String chatGuid, T item, {bool prioritized = false}) { + final queue = _queues.putIfAbsent(chatGuid, () => []); + if (queue.contains(item)) { + if (prioritized) prioritize(chatGuid, item); + return; + } + if (prioritized) { + queue.insert(0, item); + } else { + queue.add(item); + } + } + + bool remove(String chatGuid, T item) { + final queue = _queues[chatGuid]; + if (queue == null || !queue.remove(item)) return false; + if (queue.isEmpty) _queues.remove(chatGuid); + return true; + } + + bool prioritize(String chatGuid, T item) { + final queue = _queues[chatGuid]; + if (queue == null || !queue.remove(item)) return false; + queue.insert(0, item); + return true; + } + + T? next({String? activeChatGuid, required bool Function(T) isFetching}) { + final activeQueue = activeChatGuid == null ? null : _queues[activeChatGuid]; + if (activeQueue != null) { + for (final item in activeQueue) { + if (!isFetching(item)) return item; + } + } + + for (final item in all) { + if (!isFetching(item)) return item; + } + return null; + } +} diff --git a/lib/services/network/downloads_service.dart b/lib/services/network/downloads_service.dart index 8726a747ba..fbab8bee00 100644 --- a/lib/services/network/downloads_service.dart +++ b/lib/services/network/downloads_service.dart @@ -1,11 +1,10 @@ import 'package:bluebubbles/services/network/backend_service.dart'; -import 'package:bluebubbles/utils/file_utils.dart'; +import 'package:bluebubbles/services/network/attachment_download_queue.dart'; import 'package:bluebubbles/utils/logger/logger.dart'; import 'package:bluebubbles/helpers/helpers.dart'; import 'package:bluebubbles/database/models.dart'; import 'package:bluebubbles/services/services.dart'; import 'package:collection/collection.dart'; -import 'package:dio/dio.dart'; import 'package:flutter/foundation.dart'; import 'package:get/get.dart' hide Response; import 'package:path/path.dart'; @@ -18,53 +17,63 @@ AttachmentDownloadService attachmentDownloader = Get.isRegistered downloaders = [].obs; - final Map> _downloaders = {}; + final AttachmentDownloadQueue _downloaders = + AttachmentDownloadQueue(); AttachmentDownloadController? getController(String? guid) { - return _downloaders.values.flattened.firstWhereOrNull((element) => element.attachment.guid == guid); + return _downloaders.all.firstWhereOrNull((element) => element.attachment.guid == guid); } - AttachmentDownloadController startDownload(Attachment a, {Function(PlatformFile)? onComplete, Function? onError}) { + AttachmentDownloadController startDownload(Attachment a, + {Function(PlatformFile)? onComplete, Function? onError, bool prioritized = false}) { + final existing = getController(a.guid); + if (existing != null) { + if (onComplete != null && !existing.completeFuncs.contains(onComplete)) { + existing.completeFuncs.add(onComplete); + } + if (onError != null && !existing.errorFuncs.contains(onError)) { + existing.errorFuncs.add(onError); + } + if (prioritized && !existing.isFetching) prioritize(existing); + return existing; + } return Get.put(AttachmentDownloadController( attachment: a, onComplete: onComplete, onError: onError, + prioritized: prioritized, ), tag: a.guid!); } void _addToQueue(AttachmentDownloadController downloader) { downloaders.add(downloader.attachment.guid!); final chatGuid = downloader.attachment.message.target?.chat.target?.guid ?? "unknown"; - if (_downloaders.containsKey(chatGuid)) { - _downloaders[chatGuid]!.add(downloader); - } else { - _downloaders[chatGuid] = [downloader]; - } + _downloaders.add(chatGuid, downloader, prioritized: downloader.prioritized); _fetchNext(); } void _removeFromQueue(AttachmentDownloadController downloader) { downloaders.remove(downloader.attachment.guid!); final chatGuid = downloader.attachment.message.target?.chat.target?.guid ?? "unknown"; - _downloaders[chatGuid]!.removeWhere((e) => e.attachment.guid == downloader.attachment.guid); - if (_downloaders[chatGuid]!.isEmpty) _downloaders.remove(chatGuid); + _downloaders.remove(chatGuid, downloader); Get.delete(tag: downloader.attachment.guid!); _fetchNext(); } + void prioritize(AttachmentDownloadController downloader) { + if (downloader.isFetching) return; + final chatGuid = downloader.attachment.message.target?.chat.target?.guid ?? "unknown"; + _downloaders.prioritize(chatGuid, downloader); + _fetchNext(); + } + void _fetchNext() { - if (_downloaders.values.flattened.where((e) => e.isFetching).length < maxDownloads) { - AttachmentDownloadController? activeChatDownloader; - // first check if we have an active chat that needs downloads, if so prioritize that chat - if (cm.activeChat != null && _downloaders.containsKey(cm.activeChat!.chat.guid)) { - activeChatDownloader = _downloaders[cm.activeChat!.chat.guid]!.firstWhereOrNull((e) => !e.isFetching); - activeChatDownloader?.fetchAttachment(); - } - // otherwise just grab a random attachment that needs fetching - if (activeChatDownloader == null) { - _downloaders.values.flattened.firstWhereOrNull((e) => !e.isFetching)?.fetchAttachment(); - } - } + if (_downloaders.all.where((e) => e.isFetching).length >= maxDownloads) return; + final next = _downloaders.next( + activeChatGuid: cm.activeChat?.chat.guid, + isFetching: (downloader) => downloader.isFetching, + ); + next?.fetchAttachment(); } } @@ -75,6 +84,7 @@ class AttachmentDownloadController extends GetxController { final RxnNum progress = RxnNum(); final Rxn file = Rxn(); final RxBool error = RxBool(false); + final bool prioritized; Stopwatch stopwatch = Stopwatch(); bool isFetching = false; @@ -82,6 +92,7 @@ class AttachmentDownloadController extends GetxController { required this.attachment, Function(PlatformFile)? onComplete, Function? onError, + this.prioritized = false, }) { if (onComplete != null) completeFuncs.add(onComplete); if (onError != null) errorFuncs.add(onError); diff --git a/rust/src/api/api.rs b/rust/src/api/api.rs index 10a5922db2..b9b00e10a4 100644 --- a/rust/src/api/api.rs +++ b/rust/src/api/api.rs @@ -1895,11 +1895,11 @@ pub async fn download_attachment(sink: StreamSink, aps: &APSCo let mut file = std::fs::File::create(path)?; attachment.get_attachment(aps, &mut file, |prog, total| { println!("donwloading file {} of {}", prog, total); - sink.add(TransferProgress { + let _ = sink.add(TransferProgress { prog, total, attachment: None - }).unwrap(); + }); }).await?; file.flush()?; Ok(()) @@ -1914,11 +1914,11 @@ pub async fn download_mmcs(sink: StreamSink, aps: &APSConnecti let mut file = std::fs::File::create(path)?; attachment.get_attachment(aps, &mut file, |prog, total| { - sink.add(TransferProgress { + let _ = sink.add(TransferProgress { prog, total, attachment: None - }).unwrap(); + }); }).await?; file.flush()?; Ok(()) @@ -1928,7 +1928,7 @@ pub async fn download_mmcs(sink: StreamSink, aps: &APSConnecti async fn wrap_sink(sink: &StreamSink, f: impl FnOnce() -> Fut) where Fut: Future> { if let Err(err) = f().await { - sink.add_error(err).unwrap(); + let _ = sink.add_error(err); } } @@ -1945,13 +1945,13 @@ pub async fn upload_mmcs(sink: StreamSink, aps: &APSConnec let prepared = MMCSFile::prepare_put(&mut file).await?; file.rewind()?; let attachment = MMCSFile::new(aps, &prepared, file, |prog, total| { - sink.add(MMCSTransferProgress { + let _ = sink.add(MMCSTransferProgress { prog, total, file: None - }).unwrap(); + }); }).await?; - sink.add(MMCSTransferProgress { prog: 0, total: 0, file: Some(attachment) }).unwrap(); + let _ = sink.add(MMCSTransferProgress { prog: 0, total: 0, file: Some(attachment) }); Ok(()) }).await } @@ -1963,13 +1963,13 @@ pub async fn upload_attachment(sink: StreamSink, aps: &APSConn let prepared = MMCSFile::prepare_put(&mut file).await?; file.rewind()?; let attachment = Attachment::new_mmcs(aps, &prepared, file, &mime, &uti, &name,|prog, total| { - sink.add(TransferProgress { + let _ = sink.add(TransferProgress { prog, total, attachment: None - }).unwrap(); + }); }).await?; - sink.add(TransferProgress { prog: 0, total: 0, attachment: Some(attachment) }).unwrap(); + let _ = sink.add(TransferProgress { prog: 0, total: 0, attachment: Some(attachment) }); Ok(()) }).await } diff --git a/test/services/network/attachment_download_queue_test.dart b/test/services/network/attachment_download_queue_test.dart new file mode 100644 index 0000000000..f98924f8d5 --- /dev/null +++ b/test/services/network/attachment_download_queue_test.dart @@ -0,0 +1,50 @@ +import 'package:bluebubbles/services/network/attachment_download_queue.dart'; +import 'package:flutter_test/flutter_test.dart'; + +void main() { + test("a tapped attachment moves ahead of older queued work in its chat", () { + final queue = AttachmentDownloadQueue(); + queue.add("chat", "first"); + queue.add("chat", "second"); + queue.add("chat", "third"); + + expect(queue.prioritize("chat", "third"), isTrue); + expect(queue.forChat("chat"), ["third", "first", "second"]); + }); + + test("prioritizing an already queued item does not duplicate it", () { + final queue = AttachmentDownloadQueue(); + queue.add("chat", "first"); + queue.add("chat", "second"); + + queue.add("chat", "second", prioritized: true); + + expect(queue.forChat("chat"), ["second", "first"]); + }); + + test("active chat ordering wins while preserving FIFO elsewhere", () { + final queue = AttachmentDownloadQueue(); + queue.add("other", "other-1"); + queue.add("active", "active-1"); + queue.add("active", "active-2"); + + expect( + queue.next(activeChatGuid: "active", isFetching: (_) => false), + "active-1", + ); + queue.prioritize("active", "active-2"); + expect( + queue.next(activeChatGuid: "active", isFetching: (_) => false), + "active-2", + ); + }); + + test("prioritizing the first queued item preserves stable order", () { + final queue = AttachmentDownloadQueue(); + queue.add("chat", "active"); + queue.add("chat", "waiting"); + + expect(queue.prioritize("chat", "active"), isTrue); + expect(queue.forChat("chat"), ["active", "waiting"]); + }); +}