Язык программирования Rust

Стримы: последовательности футур

Вспомните, как раньше в этой главе, в разделе «Передача данных между задачами через сообщения», мы пользовались приёмником асинхронного канала. Асинхронный метод recv выдаёт последовательность элементов во времени. Это частный случай гораздо более общего шаблона — стрима (stream, асинхронного потока данных). Многое естественно представляется в виде стримов: элементы, появляющиеся в очереди; фрагменты данных, которые постепенно читаются из файловой системы, когда весь набор данных не помещается в память; данные, приходящие по сети. Раз стримы — это футуры, их можно использовать с любыми другими футурами и интересно комбинировать. Например, можно группировать события, чтобы не делать слишком много сетевых запросов, задавать таймауты для последовательностей долгих операций или притормаживать события интерфейса, чтобы не делать лишней работы.

Последовательность элементов мы уже видели в главе 13, когда разбирали трейт Iterator в разделе «Трейт Iterator и метод next». Но между итераторами и приёмником асинхронного канала два отличия. Первое — время: итераторы синхронные, а приёмник канала асинхронный. Второе — API. Работая напрямую с Iterator, мы вызываем его синхронный метод next. А у стрима trpl::Receiver мы вызывали асинхронный метод recv. В остальном эти API очень похожи, и это сходство не случайно. Стрим — это что-то вроде асинхронной формы итерации. Только trpl::Receiver специально ждёт сообщения, а универсальный API стримов гораздо шире: он выдаёт следующий элемент так же, как Iterator, но асинхронно.

Благодаря сходству итераторов и стримов в Rust стрим можно создать из любого итератора. Как и с итератором, со стримом можно работать, вызывая его метод next и дожидаясь результата, как в листинге 17-21. Правда, этот код пока не скомпилируется.

extern crate trpl; // нужно для mdbook test

fn main() {
    trpl::block_on(async {
        let values = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10];
        let iter = values.iter().map(|n| n * 2);
        let mut stream = trpl::stream_from_iter(iter);

        while let Some(value) = stream.next().await {
            println!("The value was: {value}");
        }
    });
}

Начинаем с массива чисел, превращаем его в итератор и вызываем map, чтобы удвоить все значения. Затем превращаем итератор в стрим функцией trpl::stream_from_iter. И перебираем элементы стрима по мере их поступления циклом while let.

К сожалению, при попытке запустить код он не компилируется и сообщает, что метода next нет:

error[E0599]: no method named `next` found for struct `tokio_stream::iter::Iter` in the current scope
  --> src/main.rs:10:40
   |
10 |         while let Some(value) = stream.next().await {
   |                                        ^^^^
   |
   = help: items from traits can only be used if the trait is in scope
help: the following traits which provide `next` are implemented but not in scope; perhaps you want to import one of them
   |
1  + use crate::trpl::StreamExt;
   |
1  + use futures_util::stream::stream::StreamExt;
   |
1  + use std::iter::Iterator;
   |
1  + use std::str::pattern::Searcher;
   |
help: there is a method `try_next` with a similar name
   |
10 |         while let Some(value) = stream.try_next().await {
   |                                        ~~~~~~~~

Как объясняет этот вывод, метод next станет доступен, только если подключить в область видимости соответствующий трейт. После всего сказанного логично ожидать, что это трейт Stream, но на самом деле это StreamExt. Ext — сокращение от extension (расширение); это распространённый в сообществе Rust шаблон, когда один трейт расширяют другим.

Трейт Stream задаёт низкоуровневый интерфейс, по сути объединяющий трейты Iterator и Future. А StreamExt надстраивает над Stream более высокоуровневый набор API, включая метод next и другие вспомогательные методы, похожие на те, что даёт трейт Iterator. Stream и StreamExt пока не входят в стандартную библиотеку Rust, но большинство крейтов экосистемы используют похожие определения.

Чтобы исправить ошибку компиляции, добавим инструкцию use для trpl::StreamExt, как в листинге 17-22.

extern crate trpl; // нужно для mdbook test

use trpl::StreamExt;

fn main() {
    trpl::block_on(async {
        let values = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10];
        // --пропуск--
        let iter = values.iter().map(|n| n * 2);
        let mut stream = trpl::stream_from_iter(iter);

        while let Some(value) = stream.next().await {
            println!("The value was: {value}");
        }
    });
}

Когда всё собрано вместе, код работает так, как мы хотим! Более того, теперь, когда StreamExt в области видимости, можно пользоваться всеми его вспомогательными методами — так же, как с итераторами.