生锈图案：等待溪流？

Question

在以下常见用例中构建RUST程序的正确方法是什么？

用例，

我有一个可以查询服务器的流程。这很复杂：它查询了几台服务器和交通流。数据总量太大，无法进行内存。我想将该过程的结果表示为某种数据结构，每当可能从可能交织的流中出现新的令牌时，可以等待它们。

适当的Rust/Tokio数据结构是什么？

例如，在Python/JavaScript中，这可能是异步发生器。

fn query_data(...) -> ? {
    // query servers
}

let stream = query_data(...);
async for token in stream {
    // process token
}

Answer 1

您可以从tokio-stream扩展板条箱中使用流图：

use std::time::Duration;

use tokio::time::sleep;
use tokio_stream::{wrappers::ReceiverStream, Stream, StreamExt, StreamMap};

fn create_stream(name: &'static str, delay_millis: u64) -> impl Stream<Item = String> {
    let (sender, receiver) = tokio::sync::mpsc::channel(10);

    tokio::spawn(async move {
        for i in 0..5 {
            sleep(Duration::from_millis(delay_millis)).await;
            sender.send(format!("{}-{}", name, i)).await.unwrap();
        }
    });

    let receiver_stream: ReceiverStream<String> = receiver.into();
    receiver_stream
}

fn combine_streams<T>(stream_a: T, stream_b: T) -> impl Stream<Item = (&'static str, T::Item)>
where
    T: Stream + Unpin,
{
    let mut map = StreamMap::new();
    map.insert("a", stream_a);
    map.insert("b", stream_b);
    map
}

#[tokio::main]
async fn main() {
    let stream_a = create_stream("A", 200);
    let stream_b = create_stream("B", 277);

    let mut combined = combine_streams(stream_a, stream_b);

    while let Some((origin, packet)) = combined.next().await {
        println!("Received from {origin}: {packet}");
    }
}

Received from a: A-0
Received from b: B-0
Received from a: A-1
Received from b: B-1
Received from a: A-2
Received from a: A-3
Received from b: B-2
Received from a: A-4
Received from b: B-3
Received from b: B-4