addStream method

@override Future<void> addStream(Stream<T> source, { bool? cancelOnError, })

Pipes source into this sink (events and errors), completing when the source is done. Same buffering semantics as add/addError.

The subscription is owned by this controller and cancelled by close, so a pipe left in flight at shutdown does not keep draining its source. The returned future completes on close as well as on source completion.

Implementation

@override
Future<void> addStream(Stream<T> source, {bool? cancelOnError}) {
  if (isClosed) return Future<void>.value();

  final stopOnError = cancelOnError ?? false;
  final completer = Completer<void>();
  StreamSubscription<T>? sub;

  void finish() {
    final current = sub;
    if (current != null) _pipes.removeWhere((p) => identical(p.sub, current));
    if (!completer.isCompleted) completer.complete();
  }

  final subscription = source.listen(
    add,
    onError: (Object error, StackTrace stackTrace) {
      addError(error, stackTrace);
      // cancelOnError tears the subscription down on the first error, so
      // onDone never fires. Without this the returned future would hang
      // forever and any `await sink.addStream(...)` would never return.
      if (stopOnError) finish();
    },
    onDone: finish,
    cancelOnError: stopOnError,
  );
  sub = subscription;

  // A sync source can finish inside listen(), before `sub` was assigned.
  if (completer.isCompleted) {
    unawaited(subscription.cancel());
  } else {
    _pipes.add((sub: subscription, finish: finish));
  }
  return completer.future;
}