diff --git a/packages/stream_core/CHANGELOG.md b/packages/stream_core/CHANGELOG.md index 5b2cada8..a9c799fa 100644 --- a/packages/stream_core/CHANGELOG.md +++ b/packages/stream_core/CHANGELOG.md @@ -25,7 +25,13 @@ - `Result.getOrElse`, `getOrDefault`, `recover` and `recoverCatching` return the result's own type and no longer take a type parameter. To widen, widen the result (`Result widened = intResult`) or use `fold` - Replaced the logger: `StreamLogger` is the handle you write with and a `StreamLogHandler` is where records go, so `Priority`, `MessageBuilder`, `Tag`, `IsLoggableValidator` and `Finder` are renamed or gone - `LoggingInterceptor` writes through the logger rather than printing, so it is silent until an app asks for records. Its `logPrint` is now optional, and it takes a `tag` -- The package no longer re-exports `dart:typed_data`, so code that reached `Uint8List` through `package:stream_core/stream_core.dart` must import `dart:typed_data` itself +- Reworked attachment uploads around `AttachmentUploadTask`: `StreamAttachmentUploader.upload` returns the running task rather than a `Future`, and `uploadBatch` returns an `AttachmentUploadBatch`. `CancelToken` and progress callbacks are gone from the public API +- Removed `StreamAttachment.uploadState`. Where an upload has got to lives on the task running it, not on the attachment +- Replaced the `UploadState*` classes with `UploadQueued`, `UploadPreparing`, `UploadInProgress`, `UploadSuccess`, `UploadFailed` and `UploadCancelled`, constructed directly rather than through named constructors. `UploadInProgress.progress` is an `UploadProgress` in bytes rather than a `double`, and `UploadFailed.error` is a `StreamException` +- Removed `AttachmentUploadException`. A failed upload carries the `StreamException` that stopped it, and a cancelled one a `StreamNetworkException` with `isCancelled` set +- Removed the `OnUploadProgress` and `OnBatchUploadProgress` callbacks along with the `StreamAttachmentUploaderBatch` extension. Progress arrives on `AttachmentUploadTask.state` and `AttachmentUploadBatch.state`, so it can never disagree with the lifecycle +- `uploadBatch`'s `maxConcurrent` now defaults to `3` rather than `5` +- `UploadedAttachment` compares by value rather than by identity ### ✨ Features @@ -49,6 +55,8 @@ - Added `ConnectUserDetailsRequest.fromUser`, which builds the details a client may send from a `User` - Added `StreamWebSocketClient.dispose`, which closes the connection along with `events` and `connectionState`; the client is now `Disposable`, and `connect` throws a `StateError` afterwards - Added `InFlightCache`, which hands concurrent callers asking for the same key the one call already in flight, and its outcome, success or failure alike +- Added `AttachmentUploadTask`, one upload as an object: `state` carries the whole lifecycle including byte progress, `result` settles once and never throws, and `cancel` calls it off at once +- Added `AttachmentUploadBatch`, which uploads several attachments under a concurrency limit, aggregates byte-weighted progress, and finishes as a sealed `BatchUploadResult` β€” `BatchUploadCompleted`, `BatchUploadStoppedOnError` or `BatchUploadCancelled` β€” carrying one outcome per attachment in input order ### πŸ› Bug Fixes diff --git a/packages/stream_core/lib/src/attachment.dart b/packages/stream_core/lib/src/attachment.dart index 48930047..b49fc3da 100644 --- a/packages/stream_core/lib/src/attachment.dart +++ b/packages/stream_core/lib/src/attachment.dart @@ -1,8 +1,11 @@ export 'attachment/attachment.dart'; export 'attachment/attachment_file.dart'; export 'attachment/attachment_type.dart'; -export 'attachment/attachment_upload_state.dart'; export 'attachment/cdn/cdn_client.dart'; export 'attachment/cdn/uploaded_file.dart'; +export 'attachment/uploader/attachment_upload_batch.dart'; +export 'attachment/uploader/attachment_upload_state.dart'; +export 'attachment/uploader/attachment_upload_task.dart'; export 'attachment/uploader/attachment_uploader.dart'; +export 'attachment/uploader/batch_upload_state.dart'; export 'attachment/uploader/uploaded_attachment.dart'; diff --git a/packages/stream_core/lib/src/attachment/attachment.dart b/packages/stream_core/lib/src/attachment/attachment.dart index d9f6d946..1195d8bb 100644 --- a/packages/stream_core/lib/src/attachment/attachment.dart +++ b/packages/stream_core/lib/src/attachment/attachment.dart @@ -2,13 +2,14 @@ import 'package:uuid/uuid.dart'; import 'attachment_file.dart'; import 'attachment_type.dart'; -import 'attachment_upload_state.dart'; -/// Represents a file attachment with type information and upload state. +/// Represents a file attachment with type information. /// -/// Combines an [AttachmentFile] with its [AttachmentType] and tracks the -/// upload progress through [AttachmentUploadState]. This class provides -/// a complete representation of an attachment throughout its lifecycle. +/// Combines an [AttachmentFile] with its [AttachmentType], and is what an +/// upload is asked for: it carries the local [id] the upload is addressed by, +/// the [file] to send, and the [custom] data handed back on the uploaded +/// attachment. Where the upload has got to lives on the task running it, not +/// here. /// /// Example usage: /// ```dart @@ -49,14 +50,12 @@ class StreamAttachment { /// If not provided, a UUID v4 will be automatically generated. /// The [type] specifies what kind of attachment this is. /// The [file] contains the actual file data and metadata. - /// The [uploadState] tracks the upload progress, defaulting to preparing. /// The [custom] allows storing arbitrary key-value pairs for additional /// metadata specific to your application's needs. StreamAttachment({ String? id, required this.type, required this.file, - this.uploadState = const AttachmentUploadState.preparing(), this.custom, }) : id = id ?? const Uuid().v4(); @@ -72,9 +71,6 @@ class StreamAttachment { /// The file data and metadata. final AttachmentFile file; - /// The current upload state of this attachment. - final AttachmentUploadState uploadState; - /// Optional custom data for storing arbitrary key-value pairs. /// /// This allows applications to attach additional metadata to attachments @@ -90,14 +86,12 @@ class StreamAttachment { StreamAttachment copyWith({ AttachmentType? type, AttachmentFile? file, - AttachmentUploadState? uploadState, Map? custom, }) { return StreamAttachment( id: id, // ID is preserved and cannot be changed type: type ?? this.type, file: file ?? this.file, - uploadState: uploadState ?? this.uploadState, custom: custom ?? this.custom, ); } diff --git a/packages/stream_core/lib/src/attachment/attachment_upload_state.dart b/packages/stream_core/lib/src/attachment/attachment_upload_state.dart deleted file mode 100644 index 13f51429..00000000 --- a/packages/stream_core/lib/src/attachment/attachment_upload_state.dart +++ /dev/null @@ -1,82 +0,0 @@ -/// Represents the upload state of an attachment. -/// -/// This sealed class provides a type-safe way to represent the different states -/// an attachment can be in during the upload process. Each state provides -/// relevant information for that stage of the upload. -/// -/// Example usage: -/// ```dart -/// switch (uploadState) { -/// case UploadStatePreparing(): -/// // Show preparing indicator -/// case UploadStateInProgress(:final uploaded, :final total): -/// // Show progress: uploaded/total -/// case UploadStateSuccess(): -/// // Show success state -/// case UploadStateFailed(:final error): -/// // Show error: error.toString() -/// } -/// ``` -sealed class AttachmentUploadState { - /// Creates a base [AttachmentUploadState]. - const AttachmentUploadState(); - - /// Creates a preparing state indicating upload preparation. - const factory AttachmentUploadState.preparing() = UploadStatePreparing; - - /// Creates an in-progress state with upload progress information. - const factory AttachmentUploadState.inProgress({ - required double progress, - }) = UploadStateInProgress; - - /// Creates a success state indicating successful upload completion. - const factory AttachmentUploadState.success() = UploadStateSuccess; - - /// Creates a failed state with error information. - const factory AttachmentUploadState.failed({ - required Object error, - StackTrace? stackTrace, - }) = UploadStateFailed; -} - -/// Upload state indicating the attachment is being prepared for upload. -/// -/// This is the initial state before the actual upload process begins. -class UploadStatePreparing extends AttachmentUploadState { - /// Creates a preparing state. - const UploadStatePreparing(); -} - -/// Upload state indicating the attachment upload is in progress. -/// -/// Provides progress information including bytes uploaded and total size. -class UploadStateInProgress extends AttachmentUploadState { - /// Creates an in-progress state with upload progress. - const UploadStateInProgress({required this.progress}); - - /// The upload progress as a value between 0.0 and 1.0. - final double progress; -} - -/// Upload state indicating the attachment was successfully uploaded. -class UploadStateSuccess extends AttachmentUploadState { - /// Creates a success state. - const UploadStateSuccess(); -} - -/// Upload state indicating the attachment upload failed. -/// -/// Contains error information and optionally a stack trace for debugging. -class UploadStateFailed extends AttachmentUploadState { - /// Creates a failed state with error information. - /// - /// The [error] parameter contains the error that caused the failure. - /// The [stackTrace] parameter optionally provides debugging information. - const UploadStateFailed({required this.error, this.stackTrace}); - - /// The error that caused the upload to fail. - final Object error; - - /// Optional stack trace for debugging the failure. - final StackTrace? stackTrace; -} diff --git a/packages/stream_core/lib/src/attachment/uploader/attachment_upload_batch.dart b/packages/stream_core/lib/src/attachment/uploader/attachment_upload_batch.dart new file mode 100644 index 00000000..16f1973f --- /dev/null +++ b/packages/stream_core/lib/src/attachment/uploader/attachment_upload_batch.dart @@ -0,0 +1,375 @@ +import 'dart:async'; +import 'dart:collection'; + +import 'package:meta/meta.dart'; +import 'package:uuid/uuid.dart'; + +import '../../errors/stream_exception.dart'; +import '../../utils.dart'; +import '../attachment.dart'; +import '../cdn/cdn_client.dart'; +import 'attachment_upload_state.dart'; +import 'attachment_upload_task.dart'; +import 'attachment_uploader.dart'; +import 'batch_upload_state.dart'; +import 'uploaded_attachment.dart'; + +/// Several attachment uploads run as one operation. +/// +/// A batch orchestrates [AttachmentUploadTask]s; it does not upload anything +/// itself. It decides only what runs when and when the whole thing is done β€” +/// everything per-attachment is reached through the task that owns it, so +/// cancelling, watching or awaiting one attachment is the same API whether or +/// not it is part of a batch: +/// +/// ```dart +/// final batch = uploader.uploadBatch(attachments); +/// +/// batch.state.listen((state) => showProgress(state.progress)); +/// batch.task('video-1')?.cancel(); +/// +/// switch (await batch.result) { +/// case BatchUploadCompleted(:final items): +/// submit(items); +/// case BatchUploadStoppedOnError(:final error): +/// report(error); +/// case BatchUploadCancelled(): +/// break; +/// } +/// ``` +/// +/// One upload failing does not fail the batch: a batch of three where the +/// middle one was refused still ran as asked, and finishes as +/// [BatchUploadCompleted] with that failure on its own item. Only +/// `eagerError` changes that, and only for the first failure. +/// +/// Obtained from [StreamAttachmentUploader.uploadBatch] rather than +/// constructed. Nothing needs disposing: [state] settles and stops once the +/// batch has finished, and [cancel] is how a batch is stopped before then β€” +/// including a batch abandoned while a [CdnClient] never answers, which +/// finishes on nothing else. +/// +/// See also: +/// +/// * [AttachmentUploadTask], the upload a batch is built out of. +/// * [BatchUploadState], the states a batch moves through. +/// * [BatchUploadResult], the three ways a batch can end. +abstract interface class AttachmentUploadBatch { + /// This batch's identifier, unique among the batches this process makes. + /// + /// Assigned when the batch is created and meaningful only within the process + /// that made it, so it serves to tell two batches apart in log records + /// rather than to address anything. + String get id; + + /// This batch's live state, carrying its aggregate progress. + /// + /// The current state is always available synchronously, and is the first + /// thing a new listener is given. No state follows [BatchFinished]. + StateEmitter get state; + + /// The tasks this batch orchestrates, in the order the attachments were + /// given. + /// + /// Fixed when the batch is created and unmodifiable: a batch never grows or + /// shrinks, so this is also the order [BatchUploadResult.items] arrives in. + List get uploads; + + /// Every attachment's outcome, once they have all settled. + /// + /// Never throws, and never fails as a whole: an upload's own failure is + /// carried by its [BatchUploadItemResult]. Completes however the batch ended, + /// cancellation included, so awaiting it is always safe. + Future get result; + + /// The task uploading the attachment with the given [id], or `null` if this + /// batch has none. + /// + /// The id is the [StreamAttachment.id] the batch was given, not this batch's + /// own [id]. + AttachmentUploadTask? task(String id); + + /// Calls off every upload that has not settled. + /// + /// Returns at once and is idempotent. A batch that has finished ignores it, + /// and so does one already giving up on a failure β€” it has called its + /// remaining uploads off, and finishes as [BatchUploadStoppedOnError] rather + /// than changing its mind. + /// + /// Otherwise uploads that already succeeded keep their outcome, the batch + /// moves to [BatchCancelling] until the rest have stopped, and it finishes as + /// [BatchUploadCancelled]. + /// + /// Cancelling is not undoing: an attachment already accepted stays where it + /// was put. + void cancel(); +} + +/// The [AttachmentUploadBatch] implementation. +@internal +final class AttachmentUploadBatchImpl implements AttachmentUploadBatch { + /// Creates an [AttachmentUploadBatchImpl] and starts scheduling. + /// + /// Throws an [ArgumentError] if two attachments share an id, which would + /// make [task] ambiguous, or if [maxConcurrent] is not positive. + factory AttachmentUploadBatchImpl({ + required Iterable attachments, + required CdnClient cdn, + int maxConcurrent = 3, + bool eagerError = false, + }) { + if (maxConcurrent <= 0) { + throw ArgumentError.value(maxConcurrent, 'maxConcurrent', 'A batch that may run no uploads would never finish'); + } + + // Read once: an `Iterable` is free to be lazy, and a batch validated over + // one pass but built from another could disagree about what is in it. + final requested = attachments.toList(); + + final ids = {}; + for (final attachment in requested) { + if (ids.add(attachment.id)) continue; + throw ArgumentError.value(attachment.id, 'attachments', 'Attachment ids must be unique within a batch'); + } + + final tasks = [ + for (final attachment in requested) AttachmentUploadTaskImpl(attachment: attachment, cdn: cdn), + ]; + + return AttachmentUploadBatchImpl._( + tasks, + maxConcurrent: maxConcurrent, + eagerError: eagerError, + ); + } + + AttachmentUploadBatchImpl._( + this._tasks, { + required this.maxConcurrent, + required this.eagerError, + }) : id = const Uuid().v4() { + for (final task in _tasks) { + _tasksById[task.id] = task; + _measure(task); + task.state.listen((state) => _onTaskState(task, state)); + unawaited(task.result.then((result) => _onTaskSettled(task, result))); + } + + scheduleMicrotask(_pump); + } + + @override + final String id; + + /// The most uploads allowed to be in flight at once. + final int maxConcurrent; + + /// Whether the batch gives up on the first failure. + final bool eagerError; + + final List _tasks; + final _tasksById = {}; + + // Measured attachment lengths; membership means the length is known, so a + // zero-length attachment counts as measured while an unreadable one does not. + final _totals = {}; + + // The last byte count recorded for an upload, which only a settled one is + // read from β€” a live upload is read straight off its state. + final _sent = {}; + final _started = {}; + final _active = {}; + final _outcome = Completer(); + + late final _state = MutableStateEmitter(BatchQueued(progress: _aggregate())); + + var _settledCount = 0; + _BatchEnding? _ending; + String? _failedUploadId; + StreamException? _failureError; + StackTrace? _failureStackTrace; + var _finishing = false; + + @override + StateEmitter get state => _state; + + @override + List get uploads => UnmodifiableListView(_tasks); + + @override + Future get result => _outcome.future; + + @override + AttachmentUploadTask? task(String id) => _tasksById[id]; + + @override + void cancel() { + if (_finishing || _ending != null || _outcome.isCompleted) return; + _ending = _BatchEnding.cancelled; + _cancelUnsettled(); + _emitState(); + } + + // Fills every free slot, in input order. An upload the batch already settled + // on its behalf β€” because it was cancelled, or the batch gave up β€” reads as + // final and is skipped, so giving up needs no separate check here. + void _pump() { + if (_outcome.isCompleted) return; + + for (final task in _tasks) { + if (_active.length >= maxConcurrent) break; + if (_started.contains(task.id) || task.state.value.isFinal) continue; + _started.add(task.id); + _active.add(task.id); + task.start(); + } + + _emitState(); + unawaited(_finishIfSettled()); + } + + // A queued attachment's length is known long before its turn comes, and an + // aggregate total missing one of its terms is no total at all β€” so every + // length is read up front rather than as each upload starts. + void _measure(AttachmentUploadTaskImpl task) { + unawaited( + task.measuredLength.then((length) { + if (length == null) return; + _totals[task.id] = length; + _emitState(); + }), + ); + } + + void _onTaskState(AttachmentUploadTaskImpl task, AttachmentUploadState state) { + if (state case UploadInProgress(:final progress)) _sent[task.id] = progress.sentBytes; + _emitState(); + } + + void _onTaskSettled(AttachmentUploadTaskImpl task, Result result) { + _active.remove(task.id); + _settledCount += 1; + + // Only a failure gives up on the batch. A cancellation is a decision + // somebody already made, about one upload and no others. The state says + // which of the two it was; the result carries where it was raised. + if (task.state.value case UploadFailed(:final error)) { + _stopOnError(task.id, error, result.stackTraceOrNull()); + } + + _pump(); + } + + void _stopOnError(String uploadId, StreamException error, StackTrace? stackTrace) { + if (!eagerError) return; + if (_ending != null) return; + + _ending = _BatchEnding.stoppedOnError; + _failedUploadId = uploadId; + _failureError = error; + _failureStackTrace = stackTrace; + _cancelUnsettled(); + } + + void _cancelUnsettled() { + for (final task in _tasks) { + if (task.state.value.isFinal) continue; + task.cancel(); + } + } + + void _emitState() { + if (_finishing || _state.isClosed) return; + + final progress = _aggregate(); + _state.value = switch ((_ending, _failedUploadId, _failureError)) { + (_BatchEnding.cancelled, _, _) => BatchCancelling(progress: progress), + (_BatchEnding.stoppedOnError, final failedUploadId?, final error?) => BatchStopping( + failedUploadId: failedUploadId, + error: error, + progress: progress, + ), + _ => _started.isEmpty ? BatchQueued(progress: progress) : BatchInProgress(progress: progress), + }; + } + + Future _finishIfSettled() async { + if (_finishing || _outcome.isCompleted) return; + + // Counted rather than read off the tasks: they all reach a terminal state + // before the batch is told about any of them, so a batch that waited only + // for the states would finish before it had seen the failure that stopped + // it, and call itself completed. + if (_settledCount < _tasks.length) return; + _finishing = true; + + // Read before the suspension below: a `cancel()` can still land while this + // is waiting, and a batch whose uploads all succeeded did not end up + // cancelled. + final ending = _ending; + final failureError = _failureError; + final failureStackTrace = _failureStackTrace; + final progress = _aggregate(); + + // Every task is terminal, so every outcome is already there; awaiting them + // is how the batch reads them without restating how a task settles. + final results = await Future.wait(_tasks.map((task) => task.result)); + final items = List.unmodifiable([ + for (final (index, task) in _tasks.indexed) + BatchUploadItemResult(attachment: task.attachment, result: results[index]), + ]); + + final result = switch (ending) { + _BatchEnding.stoppedOnError => BatchUploadStoppedOnError( + items: items, + error: failureError!, + stackTrace: failureStackTrace, + ), + _BatchEnding.cancelled => BatchUploadCancelled(items: items), + null => BatchUploadCompleted(items: items), + }; + + _state.value = BatchFinished(result: result, progress: progress); + await _state.close(); + + _outcome.complete(result); + } + + BatchUploadProgress _aggregate() { + final states = [for (final task in _tasks) task.state.value]; + + // Bytes and counts come from one clock: every count is derived from the + // state it is reported beside. A progress event still queued behind a + // settle would otherwise leave a finished batch reporting a fraction of + // the bytes it sent. + var sentBytes = 0; + for (final task in _tasks) { + sentBytes += switch (task.state.value) { + UploadInProgress(:final progress) => progress.sentBytes, + UploadSuccess() => _totals[task.id] ?? _sent[task.id] ?? 0, + _ => _sent[task.id] ?? 0, + }; + } + + // An attachment whose length could not be read contributes no term, and a + // total missing one of its terms would understate the work left. + final totalKnown = _totals.length == _tasks.length; + final totalBytes = _totals.values.fold(0, (total, it) => total + it); + + return BatchUploadProgress( + total: _tasks.length, + queued: states.whereType().length, + preparing: states.whereType().length, + uploading: states.whereType().length, + succeeded: states.whereType().length, + failed: states.whereType().length, + cancelled: states.whereType().length, + sentBytes: sentBytes, + totalBytes: totalKnown ? totalBytes : null, + ); + } +} + +// Which of the two deliberate endings a batch is heading for, held while it +// winds down and read once to build the result it finishes with. +enum _BatchEnding { stoppedOnError, cancelled } diff --git a/packages/stream_core/lib/src/attachment/uploader/attachment_upload_state.dart b/packages/stream_core/lib/src/attachment/uploader/attachment_upload_state.dart new file mode 100644 index 00000000..b5ec7701 --- /dev/null +++ b/packages/stream_core/lib/src/attachment/uploader/attachment_upload_state.dart @@ -0,0 +1,157 @@ +import 'package:equatable/equatable.dart'; + +import '../../errors/stream_exception.dart'; +import 'batch_upload_state.dart'; +import 'uploaded_attachment.dart'; + +/// The live state of one attachment upload. +/// +/// This is the upload's single canonical channel: progress is part of the +/// state rather than a source of its own, so there is never a moment where a +/// progress update and a lifecycle update disagree. +/// +/// ```dart +/// task.state.listen((state) { +/// switch (state) { +/// case UploadQueued(): +/// break; +/// case UploadPreparing(): +/// showPreparing(); +/// case UploadInProgress(:final progress): +/// updateProgress(progress.fraction ?? 0); +/// case UploadSuccess(:final attachment): +/// showUploaded(attachment); +/// case UploadFailed(:final error): +/// showRetry(error); +/// case UploadCancelled(): +/// removeAttachment(); +/// } +/// }); +/// ``` +/// +/// An upload settles on exactly one of [UploadSuccess], [UploadFailed] or +/// [UploadCancelled], and never moves again. +sealed class AttachmentUploadState extends Equatable { + /// Creates an [AttachmentUploadState]. + const AttachmentUploadState(); + + /// Whether the upload has settled, with no further state to follow. + bool get isFinal => switch (this) { + UploadQueued() || UploadPreparing() || UploadInProgress() => false, + UploadSuccess() || UploadFailed() || UploadCancelled() => true, + }; + + @override + List get props => const []; +} + +/// The upload is waiting for a turn, and has not started sending. +final class UploadQueued extends AttachmentUploadState { + /// Creates an [UploadQueued] state. + const UploadQueued(); +} + +/// The upload is reading the file it is about to send. +final class UploadPreparing extends AttachmentUploadState { + /// Creates an [UploadPreparing] state. + const UploadPreparing(); +} + +/// The upload is on its way, [progress] bytes of the file sent so far. +final class UploadInProgress extends AttachmentUploadState { + /// Creates an [UploadInProgress] state. + const UploadInProgress({required this.progress}); + + /// How far the upload has got. + final UploadProgress progress; + + @override + List get props => [progress]; +} + +/// The upload made it, [attachment] carrying its remote urls. +final class UploadSuccess extends AttachmentUploadState { + /// Creates an [UploadSuccess] state. + const UploadSuccess({required this.attachment}); + + /// The uploaded attachment. + final UploadedAttachment attachment; + + @override + List get props => [attachment]; +} + +/// The upload failed with [error]. +/// +/// The one state that offers the caller a retry, and the only one that +/// triggers a batch that gives up on the first failure β€” a cancellation is a +/// decision, not a failure. +final class UploadFailed extends AttachmentUploadState { + /// Creates an [UploadFailed] state. + const UploadFailed({required this.error}); + + /// What went wrong. + final StreamException error; + + @override + List get props => [error]; +} + +/// The upload was called off. +/// +/// The terminal `Result` says the same thing the rest of the SDK says about a +/// call the caller stopped: a failure carrying +/// [StreamNetworkException.isCancelled]. +final class UploadCancelled extends AttachmentUploadState { + /// Creates an [UploadCancelled] state. + const UploadCancelled(); +} + +/// How far one upload has got, in bytes. +/// +/// The counts are the attachment's own bytes, and nothing else is reported +/// against them. They are the source of truth and [fraction] is derived, which +/// is what lets a batch aggregate them; see [BatchUploadProgress.fraction]. +/// +/// A file whose length could not be read reports how much of it has gone +/// without a total to measure against, so a caller drawing a bar handles the +/// indeterminate case: +/// +/// ```dart +/// final label = switch (progress.fraction) { +/// null => 'Uploading ${progress.sentBytes} bytes…', +/// final fraction => '${(fraction * 100).round()}%', +/// }; +/// ``` +final class UploadProgress extends Equatable { + /// Creates an [UploadProgress]. + const UploadProgress({ + required this.sentBytes, + required this.totalBytes, + }); + + /// An upload that has not sent anything yet, of a file [totalBytes] long. + const UploadProgress.none({this.totalBytes}) : sentBytes = 0; + + /// The number of bytes sent so far. + final int sentBytes; + + /// The number of bytes to send, or `null` when the file's length could not + /// be read. + final int? totalBytes; + + /// The sent fraction, between 0.0 and 1.0, or `null` when [totalBytes] is + /// unknown. + /// + /// `1.0` for a file with nothing to send, which is fully sent the moment it + /// starts. + double? get fraction { + final total = totalBytes; + if (total == null) return null; + if (total == 0) return 1; + return (sentBytes / total).clamp(0.0, 1.0); + } + + @override + List get props => [sentBytes, totalBytes]; +} diff --git a/packages/stream_core/lib/src/attachment/uploader/attachment_upload_task.dart b/packages/stream_core/lib/src/attachment/uploader/attachment_upload_task.dart new file mode 100644 index 00000000..40cfdf84 --- /dev/null +++ b/packages/stream_core/lib/src/attachment/uploader/attachment_upload_task.dart @@ -0,0 +1,256 @@ +import 'dart:async'; + +import 'package:dio/dio.dart' show CancelToken; +import 'package:meta/meta.dart'; + +import '../../errors/stream_exception.dart'; +import '../../utils.dart'; +import '../attachment.dart'; +import '../attachment_type.dart'; +import '../cdn/cdn_client.dart'; +import '../cdn/uploaded_file.dart'; +import 'attachment_upload_batch.dart'; +import 'attachment_upload_state.dart'; +import 'attachment_uploader.dart'; +import 'uploaded_attachment.dart'; + +/// One attachment upload, as a handle on the operation itself. +/// +/// The upload is already running: the lifecycle is on [state], the outcome on +/// [result], and [cancel] calls it off. Watching is optional β€” a task that +/// nobody listens to runs to completion just the same. +/// +/// ```dart +/// final task = uploader.upload(attachment); +/// +/// task.state.listen((state) { +/// if (state case UploadInProgress(progress: UploadProgress(:final fraction?))) { +/// showProgress(fraction); +/// } +/// }); +/// +/// final result = await task.result; +/// result.fold( +/// onSuccess: submit, +/// onFailure: (error, _) => showRetry(error), +/// ); +/// ``` +/// +/// Obtained from [StreamAttachmentUploader.upload], or from an +/// [AttachmentUploadBatch] through [AttachmentUploadBatch.task]. Nothing needs +/// disposing: [state] settles and stops once the upload has. +/// +/// A task runs once and is never reset. Retrying means asking the uploader for +/// a new one, which is why [attachment] is kept. +/// +/// See also: +/// +/// * [AttachmentUploadBatch], which orchestrates several of these as one +/// operation. +/// * [AttachmentUploadState], the states an upload moves through. +abstract interface class AttachmentUploadTask { + /// The local identity of the attachment being uploaded, and the key this + /// upload is addressed by inside a batch. + String get id; + + /// The attachment this task uploads. + /// + /// Kept for correlation and for retrying: a failed task carries everything + /// needed to start another one. + StreamAttachment get attachment; + + /// This upload's live state, its single canonical channel. + /// + /// Progress is part of the state rather than a source of its own, so a + /// progress update and a lifecycle update can never disagree. The current + /// state is always available synchronously, and is the first thing a new + /// listener is given. + /// + /// An upload settles on exactly one of [UploadSuccess], [UploadFailed] or + /// [UploadCancelled], delivered as a value β€” a failure or a cancellation + /// never arrives as an error, so there is nothing to catch here either. + StateEmitter get state; + + /// This upload's outcome, always a value and never an error. + /// + /// A cancelled upload settles as a failure carrying a + /// [StreamNetworkException] with [StreamNetworkException.isCancelled] set, + /// the same shape every cancelled call in the SDK reports. A `Result` rather + /// than a sealed outcome like [AttachmentUploadBatch.result], because one + /// upload either produced an attachment or did not, and the reason it did + /// not is a `StreamException` the caller already knows how to read. + Future> get result; + + /// Calls the upload off. + /// + /// Returns at once, is idempotent, and is safe on a settled task, which + /// ignores it. Any other task settles as [UploadCancelled] straight away, + /// without waiting to hear what became of the upload. + /// + /// Cancelling is not undoing: an answer that arrives afterwards is dropped, + /// so an upload that had already been accepted keeps whatever it stored. + void cancel(); +} + +/// The [AttachmentUploadTask] implementation. +@internal +final class AttachmentUploadTaskImpl implements AttachmentUploadTask { + /// Creates an [AttachmentUploadTaskImpl] for [attachment], queued. + AttachmentUploadTaskImpl({ + required this.attachment, + required this._cdn, + }); + + @override + final StreamAttachment attachment; + + final CdnClient _cdn; + final _cancelToken = CancelToken(); + final _outcome = Completer>(); + final _state = MutableStateEmitter(const UploadQueued()); + + var _started = false; + + // Read once and shared with the batch, which needs every length up front to + // aggregate progress β€” without this the file would be measured twice. + late final Future _measuredLength = runSafely(() => attachment.file.size).then((it) => it.getOrNull()); + + /// The attachment's length in bytes, or `null` if it could not be read. + Future get measuredLength => _measuredLength; + + @override + String get id => attachment.id; + + @override + StateEmitter get state => _state; + + @override + Future> get result => _outcome.future; + + /// Starts the upload, unless it was already started or already settled. + void start() { + if (_started) return; + _started = true; + if (_outcome.isCompleted) return; + scheduleMicrotask(_run); + } + + @override + void cancel() { + if (_outcome.isCompleted) return; + + // The transport is told, but not waited on: a CDN client that ignores the + // token or never answers must not leave this upload β€” or a batch waiting + // on it β€” unable to settle. + _stopSending(); + _settleCancelled(); + } + + Future _run() async { + if (_outcome.isCompleted) return; + _state.value = const UploadPreparing(); + + final totalBytes = await _measuredLength; + if (_outcome.isCompleted) return; + _state.value = UploadInProgress(progress: UploadProgress.none(totalBytes: totalBytes)); + + final send = switch (attachment.type) { + AttachmentType.image => _cdn.uploadImage, + _ => _cdn.uploadFile, + }; + + final uploaded = await runSafely( + () => send( + attachment.file, + cancelToken: _cancelToken, + onProgress: (sent, total) => _trackProgress(sent: sent, wireTotal: total, fileBytes: totalBytes), + ), + ).then((it) => it.flatten()); + + uploaded.fold( + onSuccess: (file) => _settleSuccess( + UploadedAttachment( + id: attachment.id, + type: attachment.type, + custom: attachment.custom, + remoteUrl: file.fileUrl, + thumbnailUrl: file.thumbUrl, + ), + ), + onFailure: _settleFailure, + ); + } + + // The transport counts the multipart framing around the file as well as the + // file, so its counts are scaled back to the attachment's own β€” reaching the + // file's length when the request has gone out rather than as soon as the + // bytes before the framing have. + void _trackProgress({required int sent, required int wireTotal, required int? fileBytes}) { + if (_outcome.isCompleted) return; + + if (fileBytes == null) { + // Nothing to scale to, so what went out is reported as-is and the total + // stays unknown, leaving `fraction` indeterminate rather than wrong. + _state.value = UploadInProgress(progress: UploadProgress(sentBytes: sent, totalBytes: null)); + return; + } + + final sentBytes = wireTotal > 0 ? (sent / wireTotal * fileBytes).round() : sent; + _state.value = UploadInProgress( + progress: UploadProgress(sentBytes: sentBytes.clamp(0, fileBytes), totalBytes: fileBytes), + ); + } + + void _stopSending() { + if (_cancelToken.isCancelled) return; + _cancelToken.cancel('the upload was cancelled'); + } + + void _settleSuccess(UploadedAttachment uploaded) { + _settle(UploadSuccess(attachment: uploaded), Result.success(uploaded)); + } + + void _settleFailure(Object error, StackTrace? stackTrace) { + // A token this task cancelled is the authority on why the upload stopped. + // A CDN client that honours the token but reports the abort in a shape of + // its own would otherwise read as a failure, and make a batch give up. + if (_cancelToken.isCancelled) return _settleCancelled(cause: error, stackTrace: stackTrace); + + // The CDN is a pluggable seam, so a foreign error is normalized here the + // way every boundary normalizes. + var exception = StreamException.tryFrom(error); + exception ??= StreamClientException( + message: 'The upload failed', + cause: error, + ); + + // The CDN reported the cancellation itself, in the shape a caller wants and + // carrying its own transport detail. Re-wrapping it would bury that in a + // cause to say nothing new. + if (exception case StreamNetworkException(isCancelled: true)) { + return _settle(const UploadCancelled(), Result.failure(exception, stackTrace)); + } + + _settle(UploadFailed(error: exception), Result.failure(exception, stackTrace)); + } + + void _settleCancelled({Object? cause, StackTrace? stackTrace}) { + final exception = StreamNetworkException( + message: 'The upload was cancelled', + isCancelled: true, + cause: cause, + ); + + _settle(const UploadCancelled(), Result.failure(exception, stackTrace)); + } + + // The first terminal state committed wins: a success that lands while a + // cancellation is being applied cannot un-cancel the task, and the reverse + // cannot happen either. + void _settle(AttachmentUploadState finalState, Result outcome) { + if (_outcome.isCompleted) return; + _state.value = finalState; + _state.close(); + _outcome.complete(outcome); + } +} diff --git a/packages/stream_core/lib/src/attachment/uploader/attachment_uploader.dart b/packages/stream_core/lib/src/attachment/uploader/attachment_uploader.dart index dc4724e9..33cbc62b 100644 --- a/packages/stream_core/lib/src/attachment/uploader/attachment_uploader.dart +++ b/packages/stream_core/lib/src/attachment/uploader/attachment_uploader.dart @@ -1,168 +1,95 @@ -import 'dart:async'; - -import 'package:rxdart/rxdart.dart'; - -import '../../utils.dart'; import '../attachment.dart'; -import '../attachment_type.dart'; import '../cdn/cdn_client.dart'; -import 'uploaded_attachment.dart'; +import 'attachment_upload_batch.dart'; +import 'attachment_upload_task.dart'; -/// Callback for tracking upload progress. +/// An uploader of [StreamAttachment]s, sending their bytes through a +/// [CdnClient]. /// -/// Receives the upload [progress] as a value between 0.0 and 1.0. -typedef OnUploadProgress = void Function(double progress); - -/// Exception thrown when an attachment upload fails. +/// Both methods return at once, handing back the running operation rather than +/// a future to wait on: an upload has a lifecycle worth watching and a way to +/// be called off, and both belong to the object that represents it. /// -/// Provides context about which specific attachment failed and the underlying -/// cause for debugging upload issues. -class AttachmentUploadException implements Exception { - /// Creates an [AttachmentUploadException] with the specified [id] and [cause]. - const AttachmentUploadException({ - required this.id, - required this.cause, - }); - - /// The ID of the attachment that failed to upload. - final String id; - - /// The underlying cause of the upload failure. - final Object cause; - - @override - String toString() => 'AttachmentUploadException(id: $id, cause: $cause)'; -} - -/// Uploads [StreamAttachment] objects to remote storage. -/// -/// Provides upload functionality with progress tracking and error handling. -/// Automatically selects the appropriate upload method based on attachment -/// type and returns [Result] objects for explicit success/failure handling. +/// An upload that fails does not throw. It carries its error, so the outcome is +/// read rather than caught, and one attachment failing never interrupts the +/// caller. Arguments that could not describe an upload at all are the exception +/// and do throw, as documented per method. /// -/// Example usage: /// ```dart /// final uploader = StreamAttachmentUploader(cdn: cdnClient); /// -/// final result = await uploader.upload(attachment); +/// final task = uploader.upload(attachment); +/// task.state.listen(render); +/// +/// final result = await task.result; /// result.fold( /// onSuccess: (uploaded) => print('Uploaded: ${uploaded.remoteUrl}'), /// onFailure: (error, _) => print('Upload failed: $error'), /// ); /// ``` +/// +/// Where the bytes go is the [CdnClient]'s business; this tracks how far an +/// attachment has got and answers for it. Uploading somewhere else is +/// therefore a matter of supplying a different [CdnClient], not of replacing +/// this. +/// +/// Stateless, so one uploader serves any number of concurrent uploads and +/// batches; nothing is shared between them. +/// +/// See also: +/// +/// * [AttachmentUploadTask], which represents one upload. +/// * [AttachmentUploadBatch], which represents several run as one operation. +/// * [CdnClient], the seam an app supplies to upload elsewhere. class StreamAttachmentUploader { - /// Creates a [StreamAttachmentUploader] uploading through the given [CdnClient]. + /// Creates a [StreamAttachmentUploader] uploading through the given + /// [CdnClient]. const StreamAttachmentUploader({ required this._cdn, }); - // The CDN client used for upload operations. final CdnClient _cdn; - /// Uploads a single attachment to remote storage. + /// Starts uploading [attachment], and returns the task running it. /// - /// Returns a [Result] containing the [UploadedAttachment] on success or - /// an [AttachmentUploadException] on failure. Progress updates are provided - /// through the optional [onProgress] callback. - Future> upload( - StreamAttachment attachment, { - OnUploadProgress? onProgress, - }) async { - final uploadFn = switch (attachment.type) { - AttachmentType.image => _cdn.uploadImage, - _ => _cdn.uploadFile, - }; - - final result = await uploadFn( - attachment.file, - onProgress: onProgress?.let( - (f) => (uploaded, total) { - if (total == 0) return f(0); - final progress = uploaded / total; - return f(progress.clamp(0.0, 1.0)); - }, - ), - ); - - return result.fold( - onSuccess: (data) { - final uploaded = UploadedAttachment( - id: attachment.id, - type: attachment.type, - custom: attachment.custom, - remoteUrl: data.fileUrl, - thumbnailUrl: data.thumbUrl, - ); - - return Result.success(uploaded); - }, - onFailure: (cause, stackTrace) { - final ex = AttachmentUploadException( - id: attachment.id, - cause: cause, - ); - - return Result.failure(ex, stackTrace); - }, - ); + /// The upload's whole lifecycle plays out on [AttachmentUploadTask.state], it + /// can be called off through [AttachmentUploadTask.cancel], and its outcome + /// awaited through [AttachmentUploadTask.result]. + /// + /// Each call starts a new upload and a task is never reused, which is what + /// makes retrying an attachment a matter of asking again. + AttachmentUploadTask upload( + StreamAttachment attachment, + ) { + return AttachmentUploadTaskImpl( + attachment: attachment, + cdn: _cdn, + )..start(); } -} -/// Callback for tracking batch upload progress. -/// -/// Receives the [attachmentId] and upload [progress] as a value between 0.0 and 1.0 -/// for individual attachments during batch upload. -typedef OnBatchUploadProgress = void Function(String attachmentId, double progress); - -/// Extension providing batch upload functionality for [StreamAttachmentUploader]. -/// -/// Adds reactive batch upload with controlled concurrency. Results are emitted -/// as individual uploads complete, enabling immediate UI updates and partial -/// success handling. -extension StreamAttachmentUploaderBatch on StreamAttachmentUploader { - /// Uploads multiple attachments as a stream of results. - /// - /// Processes [attachments] concurrently with [maxConcurrent] limit, emitting - /// [Result] objects as each upload completes. Progress updates are provided - /// through the optional [onProgress] callback. + /// Starts uploading every attachment in [attachments], and returns the batch + /// orchestrating them. /// - /// When [eagerError] is true, the stream throws an exception and closes - /// immediately on the first upload failure. When false (default), failed - /// uploads are emitted as [Result.failure] and processing continues. + /// At most [maxConcurrent] uploads are in flight at any moment; the rest wait + /// their turn in the order they were given. When [eagerError] is true the + /// batch gives up on the first failure, calling off the uploads that have not + /// settled and never starting the ones that have not begun; when false every + /// attachment is attempted whatever the others do. An empty batch is valid, + /// and finishes at once with no items. /// - /// Returns a [Stream] of [Result] objects in completion order, not input order. - Stream> uploadBatch( + /// Throws an [ArgumentError] if [maxConcurrent] is not greater than zero, or + /// if two attachments share an id β€” a batch addresses its uploads by id, so + /// ids must be unique within one. + AttachmentUploadBatch uploadBatch( Iterable attachments, { - OnBatchUploadProgress? onProgress, - int maxConcurrent = 5, + int maxConcurrent = 3, bool eagerError = false, - }) async* { - // Early return for empty list - if (attachments.isEmpty) return; - - // Create a stream that uploads attachments with controlled concurrency - final uploadStream = Stream.fromIterable(attachments).flatMap( + }) { + return AttachmentUploadBatchImpl( + attachments: attachments, + cdn: _cdn, maxConcurrent: maxConcurrent, - (attachment) => Stream.fromFuture( - upload( - attachment, - onProgress: onProgress?.let( - (f) => - (progress) => f(attachment.id, progress), - ), - ), - ), + eagerError: eagerError, ); - - // Yield results as they complete - await for (final result in uploadStream) { - // If eagerError is enabled, throw on first failure - if (result.exceptionOrNull() case final error? when eagerError) { - final stackTrace = result.stackTraceOrNull(); - Error.throwWithStackTrace(error, stackTrace ?? StackTrace.current); - } - - yield result; - } } } diff --git a/packages/stream_core/lib/src/attachment/uploader/batch_upload_state.dart b/packages/stream_core/lib/src/attachment/uploader/batch_upload_state.dart new file mode 100644 index 00000000..2985fa00 --- /dev/null +++ b/packages/stream_core/lib/src/attachment/uploader/batch_upload_state.dart @@ -0,0 +1,287 @@ +import 'package:equatable/equatable.dart'; + +import '../../errors/stream_exception.dart'; +import '../../utils.dart'; +import '../attachment.dart'; +import 'uploaded_attachment.dart'; + +/// The live state of a batch upload. +/// +/// Every state carries the batch's [progress], so a caller that only draws a +/// progress bar never has to match on the state at all: +/// +/// ```dart +/// batch.state.listen((state) => updateOverallProgress(state.progress)); +/// ``` +sealed class BatchUploadState extends Equatable { + /// Creates a [BatchUploadState]. + const BatchUploadState(); + + /// How far the batch has got. + BatchUploadProgress get progress; + + /// Whether the batch has stopped, with no further state to follow. + bool get isFinal => this is BatchFinished; + + @override + List get props => [progress]; +} + +/// The batch has not started any of its uploads yet. +final class BatchQueued extends BatchUploadState { + /// Creates a [BatchQueued] state. + const BatchQueued({required this.progress}); + + @override + final BatchUploadProgress progress; +} + +/// The batch is working through its uploads. +final class BatchInProgress extends BatchUploadState { + /// Creates a [BatchInProgress] state. + const BatchInProgress({required this.progress}); + + @override + final BatchUploadProgress progress; +} + +/// An upload failed under a batch that gives up on the first failure, and the batch +/// is calling off the uploads that have not settled. +final class BatchStopping extends BatchUploadState { + /// Creates a [BatchStopping] state. + const BatchStopping({ + required this.failedUploadId, + required this.error, + required this.progress, + }); + + /// The id of the upload whose failure stopped the batch. + final String failedUploadId; + + /// What went wrong with that upload. + final StreamException error; + + @override + final BatchUploadProgress progress; + + @override + List get props => [...super.props, failedUploadId, error]; +} + +/// The batch was cancelled and is waiting for its uploads to stop. +/// +/// Uploads that had already succeeded keep their outcome; the rest settle as +/// cancelled, and the batch finishes as [BatchUploadCancelled]. +final class BatchCancelling extends BatchUploadState { + /// Creates a [BatchCancelling] state. + const BatchCancelling({required this.progress}); + + @override + final BatchUploadProgress progress; +} + +/// Every upload in the batch has settled, and [result] holds one outcome per +/// requested attachment. +final class BatchFinished extends BatchUploadState { + /// Creates a [BatchFinished] state. + const BatchFinished({ + required this.result, + required this.progress, + }); + + /// Every attachment's outcome, and how the batch came to stop. + final BatchUploadResult result; + + @override + final BatchUploadProgress progress; + + @override + List get props => [...super.props, result]; +} + +/// How far a batch upload has got, both in attachments and in bytes. +/// +/// The counts and the bytes answer different questions, so both are kept: +/// "uploading 3 of 7" and "5 uploaded Β· 1 failed Β· 1 remaining" come from the +/// counts, while a progress bar comes from [fraction]. +/// +/// ```dart +/// '${progress.finished} of ${progress.total}' +/// +/// // `null` until every attachment's length is known, and for a batch with +/// // one that could never be read. +/// if (progress.fraction case final fraction?) drawBar(fraction); +/// ``` +final class BatchUploadProgress extends Equatable { + /// Creates a [BatchUploadProgress]. + const BatchUploadProgress({ + required this.total, + required this.queued, + required this.preparing, + required this.uploading, + required this.succeeded, + required this.failed, + required this.cancelled, + required this.sentBytes, + required this.totalBytes, + }); + + /// The number of attachments in the batch. + final int total; + + /// How many attachments are waiting for a turn. + final int queued; + + /// How many attachments are being read. + final int preparing; + + /// How many attachments are on their way. + final int uploading; + + /// How many attachments made it. + final int succeeded; + + /// How many attachments failed. + final int failed; + + /// How many attachments were called off. + final int cancelled; + + /// The number of bytes sent so far, across the batch. + /// + /// An attachment whose length could not be read still contributes to this, + /// so it is not a count [totalBytes] can be measured against. [totalBytes] + /// is `null` whenever a batch holds one, and [fraction] reads as unknown for + /// the whole batch while that is true. + final int sentBytes; + + /// The number of attachment bytes the batch has to send. + /// + /// `null` until every attachment's length is known β€” an attachment that has + /// not been read yet cannot contribute to the total, and a total missing + /// one of its terms would understate the work left. + final int? totalBytes; + + /// How many attachments have settled. + int get finished => succeeded + failed + cancelled; + + /// The sent fraction, between 0.0 and 1.0, or `null` while [totalBytes] is + /// still unknown. + /// + /// `1.0` for a batch with nothing to send. Byte weighted rather than count + /// weighted: a 1 MB image beside a 999 MB video is 0.1% of the batch, not + /// half of it. + double? get fraction { + final total = totalBytes; + if (total == null) return null; + if (total == 0) return 1; + return (sentBytes / total).clamp(0.0, 1.0); + } + + @override + List get props => [ + total, + queued, + preparing, + uploading, + succeeded, + failed, + cancelled, + sentBytes, + totalBytes, + ]; +} + +/// One attachment's place in a [BatchUploadResult]. +final class BatchUploadItemResult extends Equatable { + /// Creates a [BatchUploadItemResult]. + const BatchUploadItemResult({ + required this.attachment, + required this.result, + }); + + /// The attachment this outcome concerns. + final StreamAttachment attachment; + + /// The upload's own outcome, failure as data. + /// + /// A cancelled upload reads as a failure carrying a [StreamNetworkException] + /// with [StreamNetworkException.isCancelled] set. + final Result result; + + @override + List get props => [attachment.id, result]; +} + +/// One outcome per requested attachment, and how the batch came to stop. +/// +/// An upload failing does not make the batch fail β€” a batch of three where the +/// middle one was refused still ran as asked, and finishes as +/// [BatchUploadCompleted]. Each attachment's own outcome is on its +/// [BatchUploadItemResult]. +/// +/// Sealed, so a `switch` over the three ways a batch can end is exhaustive, +/// and the failure that gave up on one is only reachable once it has been +/// matched: +/// +/// ```dart +/// switch (await batch.result) { +/// case BatchUploadCompleted(:final items): +/// submit(items); +/// case BatchUploadStoppedOnError(:final error): +/// report(error); +/// case BatchUploadCancelled(): +/// break; +/// } +/// ``` +/// +/// These carry `BatchUpload` where [BatchUploadState]'s members carry only +/// `Batch` on purpose: it is what tells a reader which of the two sealed +/// families a name belongs to, and it keeps [BatchCancelling] a batch still +/// calling its uploads off rather than one letter from a result. +sealed class BatchUploadResult extends Equatable { + /// Creates a [BatchUploadResult]. + const BatchUploadResult({required this.items}); + + /// One outcome per requested attachment, in the order they were given. + final List items; + + @override + List get props => [items]; +} + +/// Every attachment reached a terminal state on its own. +/// +/// "Completed" is about the batch, not its items: one that was refused, or one +/// the caller cancelled by itself, still finishes here. Only cancelling the +/// batch, or `eagerError` giving up, ends it any other way. +final class BatchUploadCompleted extends BatchUploadResult { + /// Creates a [BatchUploadCompleted] result. + const BatchUploadCompleted({required super.items}); +} + +/// An upload failed in a batch that gives up on the first failure, and the +/// rest were called off. +final class BatchUploadStoppedOnError extends BatchUploadResult { + /// Creates a [BatchUploadStoppedOnError] result. + const BatchUploadStoppedOnError({required super.items, required this.error, this.stackTrace}); + + /// The failure that gave up on the batch. + /// + /// Worth reading rather than hunting for in [items]: the uploads this + /// failure called off report cancellations of their own, which say nothing + /// about the cause. + final StreamException error; + + /// Where [error] was raised. + final StackTrace? stackTrace; + + @override + List get props => [...super.props, error]; +} + +/// The batch was called off through `batch.cancel()`. +final class BatchUploadCancelled extends BatchUploadResult { + /// Creates a [BatchUploadCancelled] result. + const BatchUploadCancelled({required super.items}); +} diff --git a/packages/stream_core/lib/src/attachment/uploader/uploaded_attachment.dart b/packages/stream_core/lib/src/attachment/uploader/uploaded_attachment.dart index 38c6a39e..8e2372aa 100644 --- a/packages/stream_core/lib/src/attachment/uploader/uploaded_attachment.dart +++ b/packages/stream_core/lib/src/attachment/uploader/uploaded_attachment.dart @@ -1,3 +1,5 @@ +import 'package:equatable/equatable.dart'; + import '../attachment_type.dart'; /// Represents a successfully uploaded attachment. @@ -17,7 +19,7 @@ import '../attachment_type.dart'; /// custom: {'source': 'camera', 'processed': true}, /// ); /// ``` -class UploadedAttachment { +class UploadedAttachment extends Equatable { /// Creates an [UploadedAttachment] with attachment details and remote URLs. const UploadedAttachment({ required this.id, @@ -49,4 +51,7 @@ class UploadedAttachment { /// Whether this attachment has a thumbnail. bool get hasThumbnail => thumbnailUrl?.isNotEmpty ?? false; + + @override + List get props => [id, type, remoteUrl, thumbnailUrl, custom]; } diff --git a/packages/stream_core/lib/src/ws/client/engine/web_socket_engine.dart b/packages/stream_core/lib/src/ws/client/engine/web_socket_engine.dart index 386b3ea9..2d8a9e90 100644 --- a/packages/stream_core/lib/src/ws/client/engine/web_socket_engine.dart +++ b/packages/stream_core/lib/src/ws/client/engine/web_socket_engine.dart @@ -3,6 +3,7 @@ import 'dart:typed_data'; import '../../../utils.dart'; import 'web_socket_options.dart'; +export 'dart:typed_data'; export 'web_socket_options.dart'; /// Interface for WebSocket engine implementations. diff --git a/packages/stream_core/lib/stream_core.dart b/packages/stream_core/lib/stream_core.dart index 33dfe3af..9ed99243 100644 --- a/packages/stream_core/lib/stream_core.dart +++ b/packages/stream_core/lib/stream_core.dart @@ -1,7 +1,7 @@ export 'package:dio/dio.dart'; export 'src/api.dart'; -export 'src/attachment.dart'; +export 'src/attachment.dart' hide AttachmentUploadBatchImpl, AttachmentUploadTaskImpl; export 'src/errors.dart'; export 'src/logger.dart'; export 'src/platform.dart'; diff --git a/packages/stream_core/test/api/interceptors/auth_interceptor_test.dart b/packages/stream_core/test/api/interceptors/auth_interceptor_test.dart index f6c12dd9..ebfa94f9 100644 --- a/packages/stream_core/test/api/interceptors/auth_interceptor_test.dart +++ b/packages/stream_core/test/api/interceptors/auth_interceptor_test.dart @@ -1,6 +1,5 @@ import 'dart:async'; import 'dart:convert'; -import 'dart:typed_data'; import 'package:stream_core/stream_core.dart'; import 'package:test/test.dart'; diff --git a/packages/stream_core/test/api/interceptors/logging_interceptor_test.dart b/packages/stream_core/test/api/interceptors/logging_interceptor_test.dart index e93f4dce..7eed8dfb 100644 --- a/packages/stream_core/test/api/interceptors/logging_interceptor_test.dart +++ b/packages/stream_core/test/api/interceptors/logging_interceptor_test.dart @@ -1,5 +1,4 @@ import 'dart:convert'; -import 'dart:typed_data'; import 'package:stream_core/stream_core.dart'; import 'package:test/test.dart'; diff --git a/packages/stream_core/test/attachment/attachment_upload_batch_cancel_test.dart b/packages/stream_core/test/attachment/attachment_upload_batch_cancel_test.dart new file mode 100644 index 00000000..8b6583ad --- /dev/null +++ b/packages/stream_core/test/attachment/attachment_upload_batch_cancel_test.dart @@ -0,0 +1,104 @@ +import 'dart:async'; + +import 'package:stream_core/stream_core.dart'; +import 'package:test/test.dart'; + +import '../helpers/attachment.dart'; +import '../helpers/fake_cdn_client.dart'; + +void main() { + group('batch cancel', () { + test('calls off every unfinished upload and keeps the finished ones', () async { + final cdn = FakeCdnClient(); + final uploader = StreamAttachmentUploader(cdn: cdn); + final attachments = attachmentsOf(4); + + final batch = uploader.uploadBatch(attachments, maxConcurrent: 2); + await pumpEventQueue(); + + cdn.upload(attachments[0]).succeed(); + await pumpEventQueue(); + + final states = []; + batch.state.listen(states.add); + + batch + ..cancel() + ..cancel(); + + final result = await batch.result; + await pumpEventQueue(); + + expect(states.whereType(), isNotEmpty); + expect(states.last, isA(), reason: 'it waits for its children before finishing'); + expect(result, isA()); + expect(result.items.map((it) => it.result.isSuccess), [true, false, false, false]); + expect(batch.uploads.skip(1).map((it) => it.state.value), everyElement(const UploadCancelled())); + expect(batch.state.isClosed, isTrue); + }); + + test('finishes even when a CDN never answers the uploads it called off', () async { + final cdn = FakeCdnClient(honoursCancellation: false); + final uploader = StreamAttachmentUploader(cdn: cdn); + final attachments = attachmentsOf(3); + + final batch = uploader.uploadBatch(attachments); + await pumpEventQueue(); + batch.cancel(); + + final result = await batch.result; + + expect(result, isA()); + expect(batch.state.isClosed, isTrue); + }); + + test('starts nothing when it arrives before the first upload begins', () async { + final cdn = FakeCdnClient(); + final uploader = StreamAttachmentUploader(cdn: cdn); + + final batch = uploader.uploadBatch(attachmentsOf(3))..cancel(); + final result = await batch.result; + + expect(cdn.received, isEmpty, reason: 'the scheduled pump must not start a cancelled batch'); + expect(result, isA()); + expect(result.items.map((it) => it.result.isSuccess), everyElement(isFalse)); + }); + + test('does not rewrite the outcome when it lands while the batch is finishing', () async { + final cdn = FakeCdnClient(); + final uploader = StreamAttachmentUploader(cdn: cdn); + final attachment = attachmentOf('a-0'); + + final batch = uploader.uploadBatch([attachment]); + await pumpEventQueue(); + + // Cancelling from the task's own outcome lands while the batch is still + // assembling its result. + unawaited(batch.uploads.single.result.then((_) => batch.cancel())); + cdn.upload(attachment).succeed(); + + final result = await batch.result; + + expect(result, isA(), reason: 'every upload succeeded'); + expect(result.items.single.result, isA>()); + }); + + test('is ignored once the batch has finished', () async { + final cdn = FakeCdnClient(); + final uploader = StreamAttachmentUploader(cdn: cdn); + final attachments = attachmentsOf(2); + + final batch = uploader.uploadBatch(attachments); + await pumpEventQueue(); + cdn.upload(attachments[0]).succeed(); + cdn.upload(attachments[1]).succeed(); + + final result = await batch.result; + batch.cancel(); + await pumpEventQueue(); + + expect(result, isA()); + expect(batch.state.value, isA()); + }); + }); +} diff --git a/packages/stream_core/test/attachment/attachment_upload_batch_error_test.dart b/packages/stream_core/test/attachment/attachment_upload_batch_error_test.dart new file mode 100644 index 00000000..ec26dd3a --- /dev/null +++ b/packages/stream_core/test/attachment/attachment_upload_batch_error_test.dart @@ -0,0 +1,214 @@ +import 'package:stream_core/stream_core.dart'; +import 'package:test/test.dart'; + +import '../helpers/attachment.dart'; +import '../helpers/fake_cdn_client.dart'; + +void main() { + group('under continueOnError', () { + test('attempts every attachment even after one fails', () async { + final cdn = FakeCdnClient(); + final uploader = StreamAttachmentUploader(cdn: cdn); + final attachments = attachmentsOf(4); + + final batch = uploader.uploadBatch(attachments, maxConcurrent: 2); + await pumpEventQueue(); + + cdn.upload(attachments[0]).succeed(); + cdn.upload(attachments[1]).fail(); + await pumpEventQueue(); + + cdn.upload(attachments[2]).succeed(); + cdn.upload(attachments[3]).succeed(); + + final result = await batch.result; + + expect(result, isA()); + expect(result.items.map((it) => it.result.isSuccess), [true, false, true, true]); + }); + + test('completes even when every upload fails', () async { + final cdn = FakeCdnClient(); + final uploader = StreamAttachmentUploader(cdn: cdn); + final attachments = attachmentsOf(3); + + final batch = uploader.uploadBatch(attachments); + await pumpEventQueue(); + for (final attachment in attachments) { + cdn.upload(attachment).fail(); + } + + final result = await batch.result; + + expect(result, isA(), reason: 'the batch ran exactly as asked'); + expect(result.items.map((it) => it.result.isSuccess), everyElement(isFalse)); + }); + }); + + group('under stopOnFirstError', () { + test('starts nothing new and calls off the rest when an upload fails', () async { + final cdn = FakeCdnClient(); + final uploader = StreamAttachmentUploader(cdn: cdn); + final attachments = attachmentsOf(5); + + final batch = uploader.uploadBatch( + attachments, + maxConcurrent: 2, + eagerError: true, + ); + + await pumpEventQueue(); + cdn.upload(attachments[0]).succeed(); + await pumpEventQueue(); + expect(cdn.received, hasLength(3), reason: 'a-2 took the freed slot'); + + cdn.upload(attachments[1]).fail(); + final result = await batch.result; + + expect(result, isA()); + expect(result.items.map((it) => it.result.isSuccess), [true, false, false, false, false]); + expect(cdn.wasReceived(attachments[3]), isFalse, reason: 'a queued upload never starts'); + expect(cdn.wasReceived(attachments[4]), isFalse); + expect(batch.uploads.skip(2).map((it) => it.state.value), everyElement(const UploadCancelled())); + }); + + test('names the upload that stopped the batch, and what went wrong', () async { + final cdn = FakeCdnClient(); + final uploader = StreamAttachmentUploader(cdn: cdn); + final attachments = attachmentsOf(3); + + final batch = uploader.uploadBatch( + attachments, + eagerError: true, + ); + + final states = []; + batch.state.listen(states.add); + + await pumpEventQueue(); + cdn.upload(attachments[1]).fail(const StreamApiException(message: 'Payload too large', statusCode: 413)); + await batch.result; + await pumpEventQueue(); + + expect( + states.whereType().first, + isA() + .having((it) => it.failedUploadId, 'failedUploadId', 'a-1') + .having( + (it) => it.error, + 'error', + isA().having((it) => it.statusCode, 'statusCode', 413), + ), + ); + }); + + test("does not give up when a cancelled upload fails in the CDN's own shape", () async { + final cdn = FakeCdnClient(); + final uploader = StreamAttachmentUploader(cdn: cdn); + final attachments = attachmentsOf(3); + + final batch = uploader.uploadBatch( + attachments, + eagerError: true, + ); + + await pumpEventQueue(); + batch.task('a-1')?.cancel(); + cdn.upload(attachments[1]).fail(StateError('connection aborted')); + await pumpEventQueue(); + + cdn.upload(attachments[0]).succeed(); + cdn.upload(attachments[2]).succeed(); + + final result = await batch.result; + + expect(result, isA()); + expect(result.items.map((it) => it.result.isSuccess), [true, false, true]); + }); + + test('carries the failure that stopped it, not the cancellations it caused', () async { + final cdn = FakeCdnClient(); + final uploader = StreamAttachmentUploader(cdn: cdn); + final attachments = attachmentsOf(4); + const refused = StreamApiException(message: 'Payload too large', statusCode: 413); + + final batch = uploader.uploadBatch(attachments, maxConcurrent: 2, eagerError: true); + await pumpEventQueue(); + cdn.upload(attachments[0]).succeed(); + cdn.upload(attachments[1]).fail(refused); + + final result = await batch.result; + + expect(result, isA().having((it) => it.error, 'error', same(refused))); + expect( + result.items.last.result, + isA().having( + (it) => it.error, + 'error', + isA().having((it) => it.isCancelled, 'isCancelled', isTrue), + ), + reason: 'the uploads it called off report cancellations of their own', + ); + }); + + test('carries the trace of the upload that stopped it', () async { + final cdn = FakeCdnClient(); + final uploader = StreamAttachmentUploader(cdn: cdn); + final attachments = attachmentsOf(2); + + final batch = uploader.uploadBatch(attachments, eagerError: true); + await pumpEventQueue(); + cdn.upload(attachments[1]).fail(); + + final result = await batch.result as BatchUploadStoppedOnError; + + // The exception says what went wrong; the trace beside it says where, and + // is the failing task's own rather than one made up here. + final failed = result.items[1].result as Failure; + expect(result.stackTrace, isNotNull, reason: 'two nulls would agree without a trace ever being kept'); + expect(result.stackTrace, same(failed.stackTrace)); + }); + + test('gives up even when the failure settles in the same turn as the last success', () async { + final cdn = FakeCdnClient(); + final uploader = StreamAttachmentUploader(cdn: cdn); + final attachments = attachmentsOf(2); + + final batch = uploader.uploadBatch(attachments, eagerError: true); + await pumpEventQueue(); + + // Both settle before the batch is told about either, so nothing is left + // queued to keep it from finishing early. + cdn.upload(attachments[0]).succeed(); + cdn.upload(attachments[1]).fail(); + + final result = await batch.result; + + expect(result, isA(), reason: 'the failure was seen before finishing'); + }); + + test('does not give up when one of its uploads is cancelled', () async { + final cdn = FakeCdnClient(); + final uploader = StreamAttachmentUploader(cdn: cdn); + final attachments = attachmentsOf(3); + + final batch = uploader.uploadBatch( + attachments, + eagerError: true, + ); + + await pumpEventQueue(); + batch.task('a-1')?.cancel(); + await pumpEventQueue(); + + cdn.upload(attachments[0]).succeed(); + cdn.upload(attachments[2]).succeed(); + + final result = await batch.result; + + expect(result, isA()); + expect(result.items.map((it) => it.result.isSuccess), [true, false, true]); + expect(batch.task('a-1')?.state.value, const UploadCancelled()); + }); + }); +} diff --git a/packages/stream_core/test/attachment/attachment_upload_batch_progress_test.dart b/packages/stream_core/test/attachment/attachment_upload_batch_progress_test.dart new file mode 100644 index 00000000..12f92ad4 --- /dev/null +++ b/packages/stream_core/test/attachment/attachment_upload_batch_progress_test.dart @@ -0,0 +1,195 @@ +import 'package:stream_core/stream_core.dart'; +import 'package:test/test.dart'; + +import '../helpers/attachment.dart'; +import '../helpers/fake_cdn_client.dart'; + +void main() { + group('batch progress', () { + test('reports the bytes it actually sent once every upload has finished', () async { + final cdn = FakeCdnClient(); + final uploader = StreamAttachmentUploader(cdn: cdn); + final attachments = attachmentsOf(3); + + final batch = uploader.uploadBatch(attachments); + await pumpEventQueue(); + + // A burst of progress callbacks in one turn is the ordinary tail of an + // upload; none of them may be delivered before the upload settles. + for (final attachment in attachments) { + final upload = cdn.upload(attachment); + for (var sent = 50; sent <= 1000; sent += 50) { + upload.sendBytes(sent, 1000); + } + upload.succeed(); + } + + final result = await batch.result; + final progress = batch.state.value.progress; + + expect(result, isA()); + expect(progress.succeeded, 3); + expect(progress.sentBytes, 3000, reason: 'three whole files were sent'); + expect(progress.fraction, 1.0); + }); + + test('measures a zero-length attachment rather than reading it as unknown', () async { + final cdn = FakeCdnClient(); + final uploader = StreamAttachmentUploader(cdn: cdn); + final empty = attachmentOf('empty', bytes: 0); + final sized = attachmentOf('sized'); + + final batch = uploader.uploadBatch([empty, sized]); + await pumpEventQueue(); + + expect(batch.state.value.progress.totalBytes, 1000); + expect(batch.state.value.progress.fraction, 0.0); + + batch.cancel(); + await batch.result; + }); + + test('leaves the total unknown while an attachment cannot be measured', () async { + final cdn = FakeCdnClient(); + final uploader = StreamAttachmentUploader(cdn: cdn); + final unreadable = StreamAttachment( + id: 'unreadable', + type: AttachmentType.file, + file: AttachmentFile('/nonexistent/does-not-exist.bin'), + ); + + final batch = uploader.uploadBatch([unreadable, attachmentOf('sized')]); + await pumpEventQueue(); + + expect(batch.state.value.progress.totalBytes, isNull); + expect(batch.state.value.progress.fraction, isNull); + + batch.cancel(); + await batch.result; + }); + + test('finishes with the total still unknown when an attachment could not be measured', () async { + final cdn = FakeCdnClient(); + final uploader = StreamAttachmentUploader(cdn: cdn); + final unreadable = StreamAttachment( + id: 'unreadable', + type: AttachmentType.file, + file: AttachmentFile('/nonexistent/does-not-exist.bin'), + ); + final sized = attachmentOf('sized'); + + final batch = uploader.uploadBatch([unreadable, sized]); + + (await cdn.awaitUpload(unreadable)) + ..sendBytes(400, 400) + ..succeed(); + (await cdn.awaitUpload(sized)).succeed(); + + final result = await batch.result; + final progress = batch.state.value.progress; + + expect(result, isA()); + expect(progress.succeeded, 2); + + // An attachment that could never be measured contributes no term, so the + // total stays unknown for good and the fraction never becomes a number β€” + // even though the batch completed. `sentBytes` still counts what went + // out, which for the unmeasured upload is what the transport reported. + expect(progress.totalBytes, isNull); + expect(progress.fraction, isNull); + expect(progress.sentBytes, 1400); + }); + + test('keeps the bytes a failed upload had already sent', () async { + final cdn = FakeCdnClient(); + final uploader = StreamAttachmentUploader(cdn: cdn); + final refused = attachmentOf('refused'); + final delivered = attachmentOf('delivered'); + + final batch = uploader.uploadBatch([refused, delivered]); + + (await cdn.awaitUpload(refused)) + ..sendBytes(400, 1000) + ..fail(); + (await cdn.awaitUpload(delivered)) + ..sendBytes(1000, 1000) + ..succeed(); + + final result = await batch.result; + final progress = batch.state.value.progress; + + expect(result, isA()); + expect(progress.failed, 1); + expect(progress.succeeded, 1); + + // A failed upload is the one case where the aggregate reads a byte count + // recorded by the state listener rather than one it can derive: there is + // no total to fall back on the way a success has. The partial bytes + // survive only because the progress event is delivered before the settle + // that follows it. + expect(progress.sentBytes, 1400, reason: '400 partial bytes plus a whole 1000-byte file'); + }); + + test('knows the whole batch total before every upload has started', () async { + final cdn = FakeCdnClient(); + final uploader = StreamAttachmentUploader(cdn: cdn); + final attachments = [ + attachmentOf('a-0'), + attachmentOf('a-1', bytes: 2000), + attachmentOf('a-2', bytes: 3000), + attachmentOf('a-3', bytes: 4000), + ]; + + final batch = uploader.uploadBatch(attachments, maxConcurrent: 2); + await pumpEventQueue(); + + expect(cdn.received, hasLength(2), reason: 'two are still queued'); + expect(batch.state.value.progress.totalBytes, 10000); + expect(batch.state.value.progress.fraction, 0.0); + + batch.cancel(); + await batch.result; + }); + + test('weighs progress by bytes, not by attachment count', () async { + final cdn = FakeCdnClient(); + final uploader = StreamAttachmentUploader(cdn: cdn); + final small = attachmentOf('small'); + final large = attachmentOf('large', bytes: 9000); + + final batch = uploader.uploadBatch([small, large]); + await pumpEventQueue(); + + cdn.upload(small).sendBytes(1000, 1000); + await pumpEventQueue(); + + expect(batch.state.value.progress.fraction, closeTo(0.1, 1e-9)); + expect(batch.state.value.progress.uploading, 2); + + batch.cancel(); + await batch.result; + }); + + test('keeps the bytes of an upload that has already succeeded', () async { + final cdn = FakeCdnClient(); + final uploader = StreamAttachmentUploader(cdn: cdn); + final small = attachmentOf('small'); + final large = attachmentOf('large', bytes: 9000); + + final batch = uploader.uploadBatch([small, large]); + await pumpEventQueue(); + + cdn.upload(small).succeed(); + await pumpEventQueue(); + + final progress = batch.state.value.progress; + expect(progress.succeeded, 1); + expect(progress.finished, 1); + expect(progress.total, 2); + expect(progress.sentBytes, 1000); + + batch.cancel(); + await batch.result; + }); + }); +} diff --git a/packages/stream_core/test/attachment/attachment_upload_batch_test.dart b/packages/stream_core/test/attachment/attachment_upload_batch_test.dart new file mode 100644 index 00000000..68b289cb --- /dev/null +++ b/packages/stream_core/test/attachment/attachment_upload_batch_test.dart @@ -0,0 +1,142 @@ +import 'package:stream_core/stream_core.dart'; +import 'package:test/test.dart'; + +import '../helpers/attachment.dart'; +import '../helpers/fake_cdn_client.dart'; + +void main() { + group('uploadBatch', () { + test('finishes an empty batch at once, so callers need no special case', () async { + final uploader = StreamAttachmentUploader(cdn: FakeCdnClient()); + + final batch = uploader.uploadBatch([]); + final result = await batch.result; + + expect(result, isA()); + expect(result.items, isEmpty); + expect(batch.uploads, isEmpty); + }); + + test('exposes the task uploading each attachment', () async { + final uploader = StreamAttachmentUploader(cdn: FakeCdnClient()); + + final batch = uploader.uploadBatch(attachmentsOf(3)); + + expect(batch.id, isNotEmpty); + expect(batch.uploads.map((it) => it.id), ['a-0', 'a-1', 'a-2']); + expect(batch.task('a-1'), same(batch.uploads[1])); + expect(batch.task('nope'), isNull); + + batch.cancel(); + await batch.result; + }); + + test('leaves no upload still emitting once it has finished', () async { + final cdn = FakeCdnClient(); + final uploader = StreamAttachmentUploader(cdn: cdn); + final attachments = attachmentsOf(5); + + // Two in flight and three never started, so the ones the batch settles + // on the caller's behalf are covered as well as the ones that ran. + final batch = uploader.uploadBatch(attachments, maxConcurrent: 2); + await pumpEventQueue(); + + batch.cancel(); + await batch.result; + await pumpEventQueue(); + + // The batch subscribes to every upload and cancels nothing: what ends + // those subscriptions is each upload closing its own channel as it + // settles, and the batch cannot finish until all of them have. + expect(batch.uploads.where((it) => !it.state.isClosed), isEmpty); + }); + + test('refuses attachments that share an id, which it could not address', () async { + final cdn = FakeCdnClient(); + final uploader = StreamAttachmentUploader(cdn: cdn); + + expect( + () => uploader.uploadBatch([attachmentOf('dupe'), attachmentOf('dupe')]), + throwsArgumentError, + ); + + // Refused before a single upload exists, rather than after some of them + // have already been set going. + await pumpEventQueue(); + expect(cdn.received, isEmpty); + }); + + test('refuses a concurrency limit that would start nothing', () { + final uploader = StreamAttachmentUploader(cdn: FakeCdnClient()); + + expect( + () => uploader.uploadBatch( + attachmentsOf(2), + maxConcurrent: 0, + ), + throwsArgumentError, + reason: 'a batch that may run no uploads would never finish', + ); + }); + + test('returns one outcome per attachment, in input order', () async { + final cdn = FakeCdnClient(); + final uploader = StreamAttachmentUploader(cdn: cdn); + final attachments = attachmentsOf(3); + + final batch = uploader.uploadBatch(attachments); + await pumpEventQueue(); + + cdn.upload(attachments[2]).succeed(fileUrl: 'c'); + cdn.upload(attachments[0]).succeed(fileUrl: 'a'); + cdn.upload(attachments[1]).succeed(fileUrl: 'b'); + + final result = await batch.result; + + expect(result.items.map((it) => it.attachment.id), ['a-0', 'a-1', 'a-2']); + expect(result.items.map((it) => it.result.getOrNull()?.remoteUrl), ['a', 'b', 'c']); + }); + + test('holds maxConcurrent as a strict upper bound', () async { + final cdn = FakeCdnClient(); + final uploader = StreamAttachmentUploader(cdn: cdn); + final attachments = attachmentsOf(5); + + final batch = uploader.uploadBatch(attachments, maxConcurrent: 2); + await pumpEventQueue(); + + expect(cdn.inFlight, 2); + expect(cdn.received, hasLength(2)); + + cdn.upload(attachments[0]).succeed(); + await pumpEventQueue(); + + expect(cdn.inFlight, 2, reason: 'a freed slot takes exactly one more'); + expect(cdn.received, hasLength(3)); + + batch.cancel(); + await batch.result; + }); + + test('refills every slot freed in the same turn, and no more', () async { + final cdn = FakeCdnClient(); + final uploader = StreamAttachmentUploader(cdn: cdn); + final attachments = attachmentsOf(5); + + final batch = uploader.uploadBatch(attachments, maxConcurrent: 2); + await pumpEventQueue(); + + // Both settle before the scheduler runs again, so it has two slots to + // fill at once rather than the one slot the case above frees. + cdn.upload(attachments[0]).succeed(); + cdn.upload(attachments[1]).succeed(); + await pumpEventQueue(); + + expect(cdn.inFlight, 2); + expect(cdn.received, hasLength(4), reason: 'both freed slots were taken, and the fifth waits'); + + batch.cancel(); + await batch.result; + }); + }); +} diff --git a/packages/stream_core/test/attachment/attachment_upload_task_cancel_test.dart b/packages/stream_core/test/attachment/attachment_upload_task_cancel_test.dart new file mode 100644 index 00000000..3c0ed1d1 --- /dev/null +++ b/packages/stream_core/test/attachment/attachment_upload_task_cancel_test.dart @@ -0,0 +1,211 @@ +import 'dart:async'; + +import 'package:stream_core/stream_core.dart'; +import 'package:test/test.dart'; + +import '../helpers/attachment.dart'; +import '../helpers/fake_cdn_client.dart'; + +void main() { + group('when the upload is cancelled', () { + test('settles while the file is still being read, sending nothing', () async { + final cdn = FakeCdnClient(); + final uploader = StreamAttachmentUploader(cdn: cdn); + final attachment = attachmentOf('file-1'); + + final task = uploader.upload(attachment); + final states = []; + task.state.listen(states.add); + + // Queued behind the microtask that starts the upload, so it lands while + // the task is reading its file and before the length has arrived β€” the + // one window where `UploadPreparing` is the live state. + scheduleMicrotask(task.cancel); + + final result = await task.result; + await pumpEventQueue(); + + expect(states, [const UploadQueued(), const UploadPreparing(), const UploadCancelled()]); + expect(cdn.wasReceived(attachment), isFalse, reason: 'the read was called off before any send'); + expect(result, isA()); + }); + }); + + group('cancel', () { + test('never touches the network when the upload has not started sending', () async { + final cdn = FakeCdnClient(); + final uploader = StreamAttachmentUploader(cdn: cdn); + + final task = uploader.upload(attachmentOf('file-1')); + final states = []; + task.state.listen(states.add); + task.cancel(); + + await task.result; + await pumpEventQueue(); + + expect(cdn.received, isEmpty, reason: 'nothing was ever handed to the CDN'); + expect( + states, + [const UploadQueued(), const UploadCancelled()], + reason: 'the run that was already scheduled must not emit past the terminal state', + ); + }); + + test('absorbs the cancelled answer the transport sends back afterwards', () async { + final cdn = FakeCdnClient(); + final uploader = StreamAttachmentUploader(cdn: cdn); + final attachment = attachmentOf('file-1'); + + final task = uploader.upload(attachment); + final states = []; + task.state.listen(states.add); + + (await cdn.awaitUpload(attachment)).sendBytes(400, 1000); + task.cancel(); + + final result = await task.result; + // The cancelled request comes back from the transport after the task has + // already settled on its own. + await pumpEventQueue(); + + expect(states.where((it) => it.isFinal), [const UploadCancelled()], reason: 'settles exactly once'); + expect(task.state.isClosed, isTrue); + expect( + result, + isA().having( + (it) => it.error, + 'error', + isA().having((it) => it.isCancelled, 'isCancelled', isTrue), + ), + ); + }); + + test('settles as the cancelled failure the rest of the SDK reports', () async { + final uploader = StreamAttachmentUploader(cdn: FakeCdnClient()); + + final task = uploader.upload(attachmentOf('file-1'))..cancel(); + + expect( + await task.result, + isA().having( + (it) => it.error, + 'error', + isA().having((it) => it.isCancelled, 'isCancelled', isTrue), + ), + ); + }); + + test('reads as cancelled when the CDN calls the request off itself', () async { + final cdn = FakeCdnClient(); + final uploader = StreamAttachmentUploader(cdn: cdn); + final attachment = attachmentOf('file-1'); + + final task = uploader.upload(attachment); + // No `task.cancel()`: the client cancels the token it was handed and + // reports the abort in a shape of its own. + (await cdn.awaitUpload(attachment)).abort(StateError('connection aborted')); + + final result = await task.result; + + expect(task.state.value, const UploadCancelled()); + expect( + result, + isA().having( + (it) => it.error, + 'error', + isA().having((it) => it.isCancelled, 'isCancelled', isTrue), + ), + ); + }); + + test('reports a cancellation the CDN shaped itself, rather than wrapping it again', () async { + final cdn = FakeCdnClient(); + final uploader = StreamAttachmentUploader(cdn: cdn); + final attachment = attachmentOf('file-1'); + + const reported = StreamNetworkException( + message: 'The request was cancelled', + isCancelled: true, + closeCode: 1000, + ); + + final task = uploader.upload(attachment); + // Nothing cancelled the token: the client answered with a cancellation of + // its own, already in the shape a caller wants. + (await cdn.awaitUpload(attachment)).fail(reported); + + final result = await task.result; + + expect(task.state.value, const UploadCancelled()); + // Passed through as it arrived, so its transport detail survives instead + // of being buried in the cause of a fresh exception. + expect(result, isA().having((it) => it.error, 'error', same(reported))); + }); + + test('settles without waiting for a CDN that never answers', () async { + final cdn = FakeCdnClient(honoursCancellation: false); + final uploader = StreamAttachmentUploader(cdn: cdn); + final attachment = attachmentOf('file-1'); + + final task = uploader.upload(attachment); + await cdn.awaitUpload(attachment); + task.cancel(); + + final result = await task.result; + + expect(task.state.value, const UploadCancelled()); + expect(result, isA()); + }); + + test('wins over an answer that lands after it', () async { + final cdn = FakeCdnClient(honoursCancellation: false); + final uploader = StreamAttachmentUploader(cdn: cdn); + final attachment = attachmentOf('file-1'); + + final task = uploader.upload(attachment); + await cdn.awaitUpload(attachment); + task.cancel(); + cdn.upload(attachment).succeed(); + + await pumpEventQueue(); + + expect(task.state.value, const UploadCancelled(), reason: 'the answer is dropped'); + expect(await task.result, isA()); + }); + + test('leaves the terminal state that was committed first', () async { + final cdn = FakeCdnClient(); + final uploader = StreamAttachmentUploader(cdn: cdn); + final attachment = attachmentOf('file-1'); + + final task = uploader.upload(attachment); + (await cdn.awaitUpload(attachment)).succeed(); + await task.result; + + task.cancel(); + await pumpEventQueue(); + + expect(task.state.value, isA()); + expect(await task.result, isA>()); + }); + + test('is safe to call more than once', () async { + final cdn = FakeCdnClient(); + final uploader = StreamAttachmentUploader(cdn: cdn); + final attachment = attachmentOf('file-1'); + + final task = uploader.upload(attachment); + (await cdn.awaitUpload(attachment)).sendBytes(100, 1000); + task + ..cancel() + ..cancel() + ..cancel(); + + await task.result; + await pumpEventQueue(); + + expect(task.state.value, const UploadCancelled()); + }); + }); +} diff --git a/packages/stream_core/test/attachment/attachment_upload_task_failure_test.dart b/packages/stream_core/test/attachment/attachment_upload_task_failure_test.dart new file mode 100644 index 00000000..7fb0c12d --- /dev/null +++ b/packages/stream_core/test/attachment/attachment_upload_task_failure_test.dart @@ -0,0 +1,66 @@ +import 'package:stream_core/stream_core.dart'; +import 'package:test/test.dart'; + +import '../helpers/attachment.dart'; +import '../helpers/fake_cdn_client.dart'; + +void main() { + group('when the upload fails', () { + test('settles as a failure, keeping the server refusal', () async { + final cdn = FakeCdnClient(); + final uploader = StreamAttachmentUploader(cdn: cdn); + final attachment = attachmentOf('file-1'); + + final task = uploader.upload(attachment); + (await cdn.awaitUpload(attachment)).fail( + const StreamApiException(message: 'Payload too large', statusCode: 413), + ); + + final result = await task.result; + + expect( + task.state.value, + isA().having( + (it) => it.error, + 'error', + isA().having((it) => it.statusCode, 'statusCode', 413), + ), + ); + expect(result, isA()); + }); + + test('normalizes an error a foreign CDN client reported', () async { + final cdn = FakeCdnClient(); + final uploader = StreamAttachmentUploader(cdn: cdn); + final attachment = attachmentOf('file-1'); + + final task = uploader.upload(attachment); + (await cdn.awaitUpload(attachment)).fail(ArgumentError('not a Stream failure')); + + await task.result; + + expect( + task.state.value, + isA().having((it) => it.error, 'error', isA()), + ); + }); + + test('settles when the CDN client throws instead of reporting a failure', () async { + final cdn = FakeCdnClient(); + final uploader = StreamAttachmentUploader(cdn: cdn); + final attachment = attachmentOf('file-1'); + + final task = uploader.upload(attachment); + (await cdn.awaitUpload(attachment)).crash(StateError('boom')); + + final result = await task.result; + + expect( + task.state.value, + isA().having((it) => it.error, 'error', isA()), + reason: 'a thrown error settles the task rather than escaping it', + ); + expect(result, isA()); + }); + }); +} diff --git a/packages/stream_core/test/attachment/attachment_upload_task_progress_test.dart b/packages/stream_core/test/attachment/attachment_upload_task_progress_test.dart new file mode 100644 index 00000000..fc0c18c3 --- /dev/null +++ b/packages/stream_core/test/attachment/attachment_upload_task_progress_test.dart @@ -0,0 +1,100 @@ +import 'package:stream_core/stream_core.dart'; +import 'package:test/test.dart'; + +import '../helpers/attachment.dart'; +import '../helpers/fake_cdn_client.dart'; + +void main() { + group('upload progress', () { + test('counts the attachment payload bytes, not the multipart bytes', () async { + final cdn = FakeCdnClient(); + final uploader = StreamAttachmentUploader(cdn: cdn); + final attachment = attachmentOf('file-1'); + + final task = uploader.upload(attachment); + final states = []; + task.state.listen(states.add); + + (await cdn.awaitUpload(attachment)) + ..sendBytes(200, 1400) + ..sendBytes(1400, 1400) + ..succeed(); + + await task.result; + await pumpEventQueue(); + + final progress = states.whereType().map((it) => it.progress); + expect(progress.map((it) => it.totalBytes), everyElement(1000)); + expect(progress.last, const UploadProgress(sentBytes: 1000, totalBytes: 1000)); + expect(progress.last.fraction, 1.0); + }); + + test('does not reach the file length until the request has gone out', () async { + final cdn = FakeCdnClient(); + final uploader = StreamAttachmentUploader(cdn: cdn); + final attachment = attachmentOf('file-1'); + + final task = uploader.upload(attachment); + // As many bytes as the file is long have gone out, but the multipart + // framing around it has not. + (await cdn.awaitUpload(attachment)).sendBytes(1000, 1400); + await pumpEventQueue(); + + expect( + task.state.value, + isA().having((it) => it.progress.fraction, 'fraction', lessThan(1.0)), + ); + + cdn.upload(attachment).succeed(); + await task.result; + }); + + test('leaves the total unknown when the file length cannot be read', () async { + final cdn = FakeCdnClient(); + final uploader = StreamAttachmentUploader(cdn: cdn); + final attachment = StreamAttachment( + id: 'file-1', + type: AttachmentType.file, + file: AttachmentFile('/nonexistent/does-not-exist.bin'), + ); + + final task = uploader.upload(attachment); + final states = []; + task.state.listen(states.add); + + (await cdn.awaitUpload(attachment)) + ..sendBytes(500, 2000) + ..succeed(); + + await task.result; + await pumpEventQueue(); + + final progress = states.whereType().map((it) => it.progress); + expect(progress.first, const UploadProgress(sentBytes: 0, totalBytes: null), reason: 'no length to report'); + + // What went out is still worth reporting; the total is not invented from + // the transport's own count, so the fraction reads as indeterminate + // rather than as a percentage of the wrong whole. + expect(progress.last, const UploadProgress(sentBytes: 500, totalBytes: null)); + expect(progress.last.fraction, isNull); + }); + + test('reads an empty file as fully sent rather than as an unknown length', () async { + final cdn = FakeCdnClient(); + final uploader = StreamAttachmentUploader(cdn: cdn); + final attachment = attachmentOf('file-1', bytes: 0); + + final task = uploader.upload(attachment); + final states = []; + task.state.listen(states.add); + + (await cdn.awaitUpload(attachment)).succeed(); + await task.result; + await pumpEventQueue(); + + final progress = states.whereType().map((it) => it.progress); + expect(progress.first, const UploadProgress(sentBytes: 0, totalBytes: 0)); + expect(progress.first.fraction, 1.0, reason: 'nothing to send is already sent'); + }); + }); +} diff --git a/packages/stream_core/test/attachment/attachment_upload_task_test.dart b/packages/stream_core/test/attachment/attachment_upload_task_test.dart new file mode 100644 index 00000000..da22828d --- /dev/null +++ b/packages/stream_core/test/attachment/attachment_upload_task_test.dart @@ -0,0 +1,115 @@ +import 'package:stream_core/stream_core.dart'; +import 'package:test/test.dart'; + +import '../helpers/attachment.dart'; +import '../helpers/fake_cdn_client.dart'; + +void main() { + group('upload', () { + test('walks the whole lifecycle on one channel, ending in success', () async { + final cdn = FakeCdnClient(); + final uploader = StreamAttachmentUploader(cdn: cdn); + final attachment = attachmentOf('image-1'); + + final task = uploader.upload(attachment); + final states = []; + task.state.listen(states.add); + + expect(task.state.value, const UploadQueued()); + + (await cdn.awaitUpload(attachment)) + ..sendBytes(500, 1000) + ..sendBytes(1000, 1000) + ..succeed(fileUrl: 'https://cdn.example.com/file.jpg', thumbUrl: 'https://cdn.example.com/thumb.jpg'); + + final result = await task.result; + await pumpEventQueue(); + + expect(states, [ + const UploadQueued(), + const UploadPreparing(), + const UploadInProgress(progress: UploadProgress(sentBytes: 0, totalBytes: 1000)), + const UploadInProgress(progress: UploadProgress(sentBytes: 500, totalBytes: 1000)), + const UploadInProgress(progress: UploadProgress(sentBytes: 1000, totalBytes: 1000)), + isA(), + ]); + + expect(task.state.isClosed, isTrue, reason: 'no state follows a terminal one'); + expect( + result.getOrNull(), + isA() + .having((it) => it.remoteUrl, 'remoteUrl', 'https://cdn.example.com/file.jpg') + .having((it) => it.thumbnailUrl, 'thumbnailUrl', 'https://cdn.example.com/thumb.jpg'), + ); + }); + + test('the task is addressed by the attachment it was given', () { + final uploader = StreamAttachmentUploader(cdn: FakeCdnClient()); + final attachment = attachmentOf('image-1'); + + final task = uploader.upload(attachment); + + expect(task.id, 'image-1'); + expect(task.attachment, same(attachment)); + }); + + test('sends an image through the image endpoint', () async { + final cdn = FakeCdnClient(); + final uploader = StreamAttachmentUploader(cdn: cdn); + final attachment = attachmentOf('image-1', type: AttachmentType.image); + + uploader.upload(attachment); + + expect((await cdn.awaitUpload(attachment)).isImage, isTrue); + }); + + test('sends anything else through the file endpoint', () async { + final cdn = FakeCdnClient(); + final uploader = StreamAttachmentUploader(cdn: cdn); + final attachment = attachmentOf('video-1', type: AttachmentType.video); + + uploader.upload(attachment); + + expect((await cdn.awaitUpload(attachment)).isImage, isFalse); + }); + + test('hands the attachment custom data back on the uploaded attachment', () async { + final cdn = FakeCdnClient(); + final uploader = StreamAttachmentUploader(cdn: cdn); + final attachment = attachmentOf('file-1', custom: {'source': 'camera'}); + + final task = uploader.upload(attachment); + (await cdn.awaitUpload(attachment)).succeed(); + + expect((await task.result).getOrNull()?.custom, {'source': 'camera'}); + }); + + test('starts a new upload every call, so an attachment can be retried', () async { + final cdn = FakeCdnClient(); + final uploader = StreamAttachmentUploader(cdn: cdn); + final attachment = attachmentOf('file-1'); + + final first = uploader.upload(attachment); + (await cdn.awaitUpload(attachment)).fail(); + await first.result; + + final second = uploader.upload(attachment); + (await cdn.awaitUpload(attachment)).succeed(); + + expect(second, isNot(same(first))); + expect(await second.result, isA>()); + }); + + test('replays the settled state to a listener that arrives late, then ends', () async { + final cdn = FakeCdnClient(); + final uploader = StreamAttachmentUploader(cdn: cdn); + final attachment = attachmentOf('file-1'); + + final task = uploader.upload(attachment); + (await cdn.awaitUpload(attachment)).succeed(); + await task.result; + + expect(await task.state.toList(), [isA()]); + }); + }); +} diff --git a/packages/stream_core/test/attachment/batch_upload_state_test.dart b/packages/stream_core/test/attachment/batch_upload_state_test.dart new file mode 100644 index 00000000..f1b1ce5d --- /dev/null +++ b/packages/stream_core/test/attachment/batch_upload_state_test.dart @@ -0,0 +1,71 @@ +import 'package:stream_core/stream_core.dart'; +import 'package:test/test.dart'; + +import '../helpers/attachment.dart'; + +void main() { + group('BatchUploadItemResult', () { + test('compares two outcomes for the same attachment by what they carry', () { + const uploaded = UploadedAttachment( + id: 'a-0', + type: AttachmentType.file, + remoteUrl: 'https://cdn.example.com/file', + ); + + // The same attachment handed over twice is two objects, and the file it + // wraps holds bytes no caller wants compared. Its id is what addresses + // the item, so that is what equality reads. + final first = BatchUploadItemResult( + attachment: attachmentOf('a-0'), + result: const Result.success(uploaded), + ); + final second = BatchUploadItemResult( + attachment: attachmentOf('a-0'), + result: const Result.success(uploaded), + ); + + expect(first, second); + expect(first.hashCode, second.hashCode); + }); + + test('tells two attachments apart, and two outcomes apart', () { + const uploaded = UploadedAttachment(id: 'a-0', type: AttachmentType.file); + + final item = BatchUploadItemResult( + attachment: attachmentOf('a-0'), + result: const Result.success(uploaded), + ); + final otherAttachment = BatchUploadItemResult( + attachment: attachmentOf('a-1'), + result: const Result.success(uploaded), + ); + final otherOutcome = BatchUploadItemResult( + attachment: attachmentOf('a-0'), + result: const Result.success( + UploadedAttachment(id: 'a-0', type: AttachmentType.file, remoteUrl: 'https://cdn.example.com/other'), + ), + ); + + expect(item, isNot(otherAttachment)); + expect(item, isNot(otherOutcome)); + }); + + test('compares a failed outcome by the exception it carries', () { + final refused = BatchUploadItemResult( + attachment: attachmentOf('a-0'), + result: const Result.failure(StreamApiException(message: 'Refused', statusCode: 400)), + ); + final same = BatchUploadItemResult( + attachment: attachmentOf('a-0'), + result: const Result.failure(StreamApiException(message: 'Refused', statusCode: 400)), + ); + final different = BatchUploadItemResult( + attachment: attachmentOf('a-0'), + result: const Result.failure(StreamApiException(message: 'Refused', statusCode: 403)), + ); + + expect(refused, same); + expect(refused, isNot(different)); + }); + }); +} diff --git a/packages/stream_core/test/helpers/attachment.dart b/packages/stream_core/test/helpers/attachment.dart new file mode 100644 index 00000000..c62c3f96 --- /dev/null +++ b/packages/stream_core/test/helpers/attachment.dart @@ -0,0 +1,21 @@ +import 'package:stream_core/stream_core.dart'; + +/// An attachment of exactly [bytes] bytes. +StreamAttachment attachmentOf( + String id, { + int bytes = 1000, + AttachmentType type = AttachmentType.file, + Map? custom, +}) { + return StreamAttachment( + id: id, + type: type, + file: AttachmentFile.fromData(Uint8List(bytes), name: '$id.bin'), + custom: custom, + ); +} + +/// [count] attachments named `a-0`, `a-1`, ... of [bytes] bytes each. +List attachmentsOf(int count, {int bytes = 1000}) { + return [for (var index = 0; index < count; index++) attachmentOf('a-$index', bytes: bytes)]; +} diff --git a/packages/stream_core/test/helpers/fake_cdn_client.dart b/packages/stream_core/test/helpers/fake_cdn_client.dart new file mode 100644 index 00000000..382eea5e --- /dev/null +++ b/packages/stream_core/test/helpers/fake_cdn_client.dart @@ -0,0 +1,162 @@ +import 'dart:async'; + +import 'package:stream_core/stream_core.dart'; + +/// A [CdnClient] the test drives by hand. +/// +/// Uploads park here until the test sends bytes, succeeds them or fails them, +/// which is what makes the scheduler's timing β€” who is in flight, who is still +/// queued β€” observable at all. +/// +/// Uploads are addressed by the attachment they came from rather than by file +/// name, because a file built from bytes carries no name on every platform. +class FakeCdnClient implements CdnClient { + /// Creates a [FakeCdnClient]. + /// + /// When [honoursCancellation] is false a cancelled token is ignored, the way + /// a third-party client that never answers would behave. + FakeCdnClient({this.honoursCancellation = true}); + + /// Whether a cancelled token comes back as a cancelled failure. + final bool honoursCancellation; + + final _uploads = {}; + final _awaited = >{}; + final _received = []; + + /// The attachments handed over for upload, in the order they arrived. + List get received => List.unmodifiable(_received); + + /// How many uploads are in flight right now. + int get inFlight => _uploads.values.where((it) => !it.isSettled).length; + + /// Whether [attachment] has been handed over at all. + bool wasReceived(StreamAttachment attachment) => _uploads.containsKey(attachment.file); + + /// The upload of [attachment]. + FakeUpload upload(StreamAttachment attachment) { + final upload = _uploads[attachment.file]; + if (upload != null) return upload; + throw StateError('No upload was handed over for "${attachment.id}"'); + } + + /// Completes once [attachment] has been handed over for upload. + Future awaitUpload(StreamAttachment attachment) { + final upload = _uploads[attachment.file]; + if (upload != null && !upload.isSettled) return Future.value(upload); + return (_awaited[attachment.file] ??= Completer()).future; + } + + @override + Future> uploadImage( + AttachmentFile image, { + ProgressCallback? onProgress, + CancelToken? cancelToken, + }) { + return _accept(image, isImage: true, onProgress: onProgress, cancelToken: cancelToken); + } + + @override + Future> uploadFile( + AttachmentFile file, { + ProgressCallback? onProgress, + CancelToken? cancelToken, + }) { + return _accept(file, isImage: false, onProgress: onProgress, cancelToken: cancelToken); + } + + @override + Future> deleteImage(String url, {CancelToken? cancelToken}) => throw UnimplementedError(); + + @override + Future> deleteFile(String url, {CancelToken? cancelToken}) => throw UnimplementedError(); + + Future> _accept( + AttachmentFile file, { + required bool isImage, + ProgressCallback? onProgress, + CancelToken? cancelToken, + }) { + final upload = FakeUpload._( + file: file, + isImage: isImage, + onProgress: onProgress, + cancelToken: cancelToken, + ); + _uploads[file] = upload; + _received.add(file); + _awaited.remove(file)?.complete(upload); + + // A cancelled request comes back as the transport's own cancelled failure, + // the way a Dio-backed CDN client reports one. + if (honoursCancellation) { + unawaited(cancelToken?.whenCancel.then((_) => upload._cancel())); + } + + return upload._outcome.future; + } +} + +/// One upload parked inside a [FakeCdnClient]. +class FakeUpload { + FakeUpload._({ + required this.file, + required this.isImage, + required this._onProgress, + required this._cancelToken, + }); + + /// The file the uploader handed over. + final AttachmentFile file; + + /// Whether it arrived through the image endpoint. + final bool isImage; + + final ProgressCallback? _onProgress; + final CancelToken? _cancelToken; + final _outcome = Completer>(); + + /// Whether this upload has settled. + bool get isSettled => _outcome.isCompleted; + + /// Reports [sent] of [total] bytes sent, as the transport counts them. + void sendBytes(int sent, int total) => _onProgress?.call(sent, total); + + /// Answers with the uploaded file's urls. + void succeed({String fileUrl = 'https://cdn.example.com/file', String? thumbUrl}) { + _settle(Result.success(UploadedFile(fileUrl: fileUrl, thumbUrl: thumbUrl))); + } + + /// Answers with a failure. + void fail([Object error = const StreamApiException(message: 'Refused', statusCode: 400)]) { + _settle(Result.failure(error, StackTrace.current)); + } + + /// Calls the request off on its own initiative and answers with [error], + /// the way a client that gives up without being asked to would. + void abort([Object error = 'the client gave up']) { + _cancelToken?.cancel(); + _settle(Result.failure(error, StackTrace.current)); + } + + /// Throws instead of answering, the way a CDN client that does not report + /// failure as a [Result] would. + void crash([Object error = 'the CDN client threw']) { + if (_outcome.isCompleted) return; + _outcome.completeError(error, StackTrace.current); + } + + void _cancel() { + _settle( + Result.failure( + const StreamNetworkException(message: 'The request was cancelled', isCancelled: true), + StackTrace.current, + ), + ); + } + + void _settle(Result outcome) { + if (_outcome.isCompleted) return; + _outcome.complete(outcome); + } +}