download method

Stream<PrismDownloadEvent> download({ required Uri url, required String savePath, Duration throughputWindow = const Duration(seconds: 1), })

Streams url into savePath, emitting PrismDownloadProgress events on every chunk and exactly one terminal event at the end.

Cancelling the subscription cancels the HTTP response and stops writing.

Implementation

Stream<PrismDownloadEvent> download({
  required Uri url,
  required String savePath,
  Duration throughputWindow = const Duration(seconds: 1),
}) async* {
  IOSink? sink;
  StreamSubscription<List<int>>? responseSub;

  try {
    final request = await _client.getUrl(url);
    final response = await request.close();
    if (response.statusCode < 200 || response.statusCode >= 300) {
      yield PrismDownloadFailed(
        HttpException('Download failed (${response.statusCode})', uri: url),
      );
      return;
    }

    final total = response.contentLength == -1 ? -1 : response.contentLength;
    final file = File(savePath);
    await file.parent.create(recursive: true);
    sink = file.openWrite();

    var received = 0;
    var windowBytes = 0;
    var speed = 0.0;
    final windowSw = Stopwatch()..start();
    final windowSeconds = throughputWindow.inMilliseconds / 1000;

    // Bridge response chunks through a controller so that cancelling the
    // generator's subscription cancels the underlying HTTP response.
    final chunks = StreamController<List<int>>();
    responseSub = response.listen(
      chunks.add,
      onError: chunks.addError,
      onDone: chunks.close,
      cancelOnError: true,
    );
    _active.add(responseSub);

    await for (final chunk in chunks.stream) {
      received += chunk.length;
      windowBytes += chunk.length;
      final elapsed = windowSw.elapsedMicroseconds / 1e6;
      if (elapsed >= windowSeconds && elapsed > 0) {
        speed = windowBytes / elapsed;
        windowBytes = 0;
        windowSw.reset();
      }
      sink.add(chunk);
      yield PrismDownloadProgress(
        received: received,
        total: total,
        bytesPerSecond: speed,
      );
    }

    await sink.flush();
    await sink.close();
    yield PrismDownloadComplete(file, bytes: received);
  } catch (e, st) {
    yield PrismDownloadFailed(e, st);
  } finally {
    if (responseSub != null) {
      _active.remove(responseSub);
      unawaited(responseSub.cancel());
    }
    try {
      await sink?.close();
    } catch (_) {
      // Best-effort cleanup; the terminal event already reported status.
    }
  }
}