Как отобразить () поток до его отмены в Dart?

Question

У меня есть Stream Я получаю из запроса к базе данных.Я считаю, что запрос будет выполняться в течение некоторого времени, производя значения по мере его выполнения, поэтому я хотел, чтобы у меня была возможность показать пользователю элементы, как только они станут доступны.

Но как только пользователь выбирает элемент, яхочу, чтобы поток был отменен.

У меня проблемы с написанием этого, потому что я не понимаю, как получить подписку на поток, которую можно отменить позже, и в то же время сопоставить ее элементы,так что потребитель моего сопоставленного потока может обрабатывать элементы так, как они создаются исходным потоком.

В принципе, я думаю, что мне нужно что-то вроде CancellableStream, но я не вижу ничего подобного вDart SDK.

Я уже пробовал что-то подобное:

final subscription = cursor.listen((entry) => process(entry));
// now I can cancel the subscription when needed, but how to
// return the processed items to the caller?

final processed = cursor.map((entry) => process(entry));
// now I have the processed Stream I wanted, but how can I cancel it?

Answer 1

На основе https://www.dartlang.org/articles/libraries/creating-streams...

Вот простая реализация отменяемого потока:

class CancellableStream<T> {
  final Stream<T> _originalStream;
  bool _isCancelled = false;

  CancellableStream(this._originalStream);

  Stream<T> get stream async* {
    await for (final item in _originalStream) {
      if (_isCancelled) break;
      yield item;
    }
  }

  void cancel() => _isCancelled = true;
}

Answer 2

Я думаю, where(...) с использованием состояния hasPicked должно делать то, что вы хотите

bool hasPicked = false;
...
final processed = cursor.where((entry) => !hasPicked).map((entry) => process(entry));

Когда пользователь выберет один, установите hasPicked на true.