@override Stream<T> stream() { if (_consumed) throw StateError('stream is single-subscription'); _consumed = true; return _stream(); }