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