writeStream method

@override Future<void> writeStream(String key, Stream<List<int>> byteStream, [ WriteOptions options = const WriteOptions() ])

Write a byte stream with optional write options.

Creates parent structure if needed. Replaces any existing object. Primary write path — write delegates to this.

options controls encryption, content type, and metadata.

Implementation

@override
Future<void> writeStream(
  String key,
  Stream<List<int>> byteStream, [
  WriteOptions options = const WriteOptions(),
]) async {
  checkNotDisposed();
  final previous = await _readManifest(key);
  // Same failed-first-write sweep as [write].
  if (previous == null && await _backing.exists(chunkKeyFor(key, 0, 0))) {
    await _backing.deletePrefix('$key$generationPrefix');
  }
  final generation = (previous?.generation ?? -1) + 1;
  var index = 0;
  var totalSize = 0;
  final buffer = BytesBuilder(copy: false);

  Future<void> flush() async {
    if (buffer.isEmpty) return;
    final chunk = buffer.takeBytes();
    await _putChunk(key, generation, index, chunk);
    index++;
  }

  await for (final chunk in byteStream) {
    var offset = 0;
    while (offset < chunk.length) {
      final remaining = chunkSize - buffer.length;
      final take = (chunk.length - offset) < remaining
          ? chunk.length - offset
          : remaining;
      buffer.add(chunk.sublist(offset, offset + take));
      offset += take;
      totalSize += take;
      if (buffer.length == chunkSize) await flush();
    }
    // Logical-bytes progress, once per source chunk. _putChunk builds
    // its own per-record options, so the backing never double-reports.
    options.onProgress?.call(totalSize, null);
  }
  await flush(); // final partial chunk (may be empty for empty input)

  await _writeManifest(
    key,
    generation: generation,
    chunkCount: index,
    totalSize: totalSize,
    contentType: options.contentType,
    metadata: options.metadata,
  );
  await _deleteGeneration(key, previous);
}