pipe method

Future<void> pipe(Stream<List<int>> source, StreamConsumer<List<int>> sink)

Writes source to sink through addStream, counting what sink has not taken yet.

source is never paused: dart:io pauses the stream it writes from while the kernel is full, and what it holds above a paused stream is uncounted, so every byte has to land in a queue this counts.

Implementation

Future<void> pipe(Stream<List<int>> source, StreamConsumer<List<int>> sink) {
  if (maxUnsentBytes == null) return sink.addStream(source);
  // Synchronous, so a write the kernel can take now is not held a microtask.
  final queue = StreamController<List<int>>(sync: true);
  late final StreamSubscription<List<int>> feed;
  final writing = sink.addStream(queue.stream.map(_taken)).whenComplete(() {
    unawaited(feed.cancel());
    _release();
  });
  feed = source.listen(
    (data) {
      _unsent += data.length;
      queue.add(data);
      if (!_over && _unsent > maxUnsentBytes!) {
        _over = true;
        _reader?.pause();
      }
    },
    onError: queue.addError,
    onDone: queue.close,
    cancelOnError: true,
  );
  return writing;
}