upload method

@override Future<bool> upload(Stream<List<int>> stream, int length, { UploadProgressCallback? onProgress, })

Reads stream into memory, then uploads it. Returns true when the storage finalized the file.

Throws DioException on a transfer failure that outlasts the retries, ResumableUploadException when the storage keeps answering without persisting or breaks the protocol.

Implementation

@override
Future<bool> upload(
  final Stream<List<int>> stream,
  final int length, {
  final UploadProgressCallback? onProgress,
}) async {
  if (_attemptedUpload) {
    throw StateError('Data has already been uploaded using this uploader.');
  }
  _attemptedUpload = true;

  final bytes = await _collectBytes(stream);
  if (bytes.length != length) {
    throw ArgumentError.value(
      length,
      'length',
      'does not match the ${bytes.length} bytes read from the stream',
    );
  }
  final total = bytes.length;
  final sessionUri = _sessionUri;

  var offset = 0;
  var failures = 0;
  var resumeFromStatus = false;
  while (true) {
    try {
      if (resumeFromStatus) {
        final status = await _queryStatus(sessionUri, total);
        if (status.completed) {
          onProgress?.call(total, total);
          return true;
        }
        if (status.persistedBytes > offset) {
          failures = 0;
        }
        offset = status.persistedBytes;
        resumeFromStatus = false;
      }

      final end = math.min(offset + _chunkSize, total);
      final result = await _sendChunk(
        sessionUri,
        bytes,
        start: offset,
        end: end,
        total: total,
        onProgress: onProgress,
      );
      if (result.completed) {
        onProgress?.call(total, total);
        return true;
      }
      if (result.persistedBytes <= offset) {
        if (failures >= _maxRetries) {
          throw ResumableUploadException(
            'The storage did not persist the upload chunk at offset $offset.',
          );
        }
        failures++;
        resumeFromStatus = true;
        await Future<void>.delayed(_retryDelay(failures));
        continue;
      }
      offset = result.persistedBytes;
      failures = 0;
    } on DioException catch (e) {
      if (!_isTransient(e) || failures >= _maxRetries) {
        rethrow;
      }
      failures++;
      resumeFromStatus = true;
      await Future<void>.delayed(_retryDelay(failures));
    }
  }
}