Future<T?> addStream(Stream<T> source, {bool cancelOnError = true}) { return _subject.addStream(source, cancelOnError: cancelOnError) as Future<T?>; }