Future addStream(Stream<T> source, {bool? cancelOnError}) => _controller.addStream(source, cancelOnError: cancelOnError);