Как я могу осуществлять двустороннюю связь в Dart с StreamChannel?

Question

Есть ли способ с Dart Streams сделать что-то вроде await stream.first несколько раз подряд, чтобы использовать поток, немного похожий на стек / очередь?

Насколько я понимаю, вы могли бы сделать это, используя asBroadcastStream в существующем потоке с одной подстрокой, но я чувствую, что это не идеально.

Возможно, существует некий буферный примитив скакой-то пакет Rx или что-то в этом роде?

Мой пример использования следующий:

У меня есть StreamChannel (IOWebSocketChannel), и я хочу отправлять / получать сообщения в определенном порядке.

т.е.:

Send Message0
Receive Message1
Send Message2
Receive Message3

, и я точно знаю, что Message1 поступит только после того, как сервер получит Message0 (и т. Д. Для Message2 и Message3) * * тысяча двадцать-один

Answer 1

У Dart есть преимущество (и, на мой взгляд, недостаток) в том, что у него есть одна абстракция для событий, основанных на push * и чтение ресурсов на основе pull.Это привело к множеству других недоразумений, таких как «Нужно ли отменять потоковые подписки» .

В модели на основе push-уведомлений произнесите:

abstract class Element {
  Stream<MouseEvent> get onClick;
}

Вы уведомлены классом при клике .Может быть где-то между 0 и буквально бесконечными событиями щелчка, и нетипично хотеть буферизовать или обрабатывать их так же, как при чтении ресурса (особенно если учесть, что Dart однопоточный). Еще одно примечание : вполне допустимо иметь любое число подписчиков этого потока (несколько классов могут быть заинтересованы в том, чтобы знать, когда происходит щелчок).

С другой стороныС другой стороны, существует модель, основанная на извлечении, например, для чтения файла:

abstract class File {
  Stream<String> readLines();
}

В в этом случае вы можете обрабатывать построчно, возможно, даже останавливаякак только вы нажмете определенную строку, и вы определенно захотите получить уведомление от EOF (обычно в случае события «done» или закрытия потока в идиоматическом Dart). Еще одно примечание : не действительно, если у вас более 1 подписчика - это очень сложно.

По вашему конкретному вопросу, кажется, вы хотите:

• Отправить собственное событие
• Получить следующее событие по запросу
• Обработать событие, отправить собственное событие
• и т.д ...

ОК, давайте перейдем к конкретике вашего вопроса:

---

Насколько я понимаю, вы можете сделать это, используя asBroadcastStreamв существующем потоке с одной подстрокой, но я чувствую, что это не идеально.

Очень неидеально.Потоки с одной подпиской (я называю эти «потоки ресурсов» в моей голове, как readLines вызов выше) автоматически буферизует события и ждет подписчика.Было бы очень плохо, если бы не получили строку (и) текста, потому что подписка произошла после чтения файла.

С другой стороны, трансляцияПоток не буфер событий.Таким образом, вы можете оказаться потерявшими событиями, отправленными, если вы использовали свой подход в зависимости от времени и т. Д.

Здесь есть несколько вариантов.Ни один из них не идеален, но они могут помочь:

• В пакете package:stream_transform есть набор часто используемых преобразований для Stream, включая нескольковдохновлены RX.

• В пакете package:async есть и другие утилиты для работы с асинхронным кодом.В частности, здесь вы можете найти, что StreamQueue делает именно то, что вы хотите:

  Future<void> processEvents(Stream<String> inputStream) async {
    var queue = new StreamQueue(inputStream);
    while (await queue.hasNext) {
      var next = await queue.next;
      // Insert processing here.
    }
  }
  

Я как бы желаю, чтобы StreamQueue было в dart:async, и не было общего интерфейса между Stream s, используемыми для ресурсов, и потоками, используемыми для событий, но сегодня это достойный подход.

Cheers!