futures_util::stream::Peekable 不是偷看

Question

熟悉

peekable()

和

Peek

来自

Iterator

我以为我知道

futures

版本会如何表现。

但令我惊讶的是，它不是偷看 - 它实际上是从 Stream 中删除项目，以便在我调用时它们不可用

next()

.

当我在下面的代码中多次等待

peek_first_message()

时，它会显示different websocket messages.

#[derive(Debug, Deserialize)]
struct WsMessage {
   ... // elided
}

async fn peek_first_message(
    mut read: Pin<&mut impl Stream<Item = Result<Message, Error>>>,
) -> anyhow::Result<()> {
    let read = read.peekable();
    tokio::pin!(read);
    let x = read
        .as_mut()
        .peek()
        .await
        .ok_or(anyhow::anyhow!("websocket closed before first message"))?
        .as_ref();
    let ws: WsMessage = match x {
        Ok(v) => { serde_json::from_slice(v.to_text()?.as_bytes())? },
        Err(e) => { return Err(anyhow::anyhow!("fail")) },
    };
    println!("{ws:?}");
    return Ok(())
}

#[tokio::main]
async fn main() {
    let url = url::Url::parse("wss://127.0.0.1:12345").unwrap();
    let (ws_stream, _) = tokio_tungstenite::connect_async(url).await.expect("Failed to connect");
    let (_, read) = ws_stream.split();
    tokio::pin!(read);

    peek_first_message(read.as_mut()).await;
    peek_first_message(read.as_mut()).await;
    peek_first_message(read.as_mut()).await;
}

我在

peek

中只有

peek_first_message

但是当我检查打印语句时，我发现

process_messages

函数没有打印出偷看的消息。这就像我刚刚在 peeking 函数中调用了

next()

一样。这是怎么回事？

Answer 1

这正是预期的行为，因为您在函数中创建了一个新的

Peekable

，它必须每次调用底层迭代器的

next

来填充它的缓冲区。

Peekable

在迭代器上调用

next

如果它的缓冲区是空的.

如果你想保留缓冲区，你必须传递

Peekable

，因为它包含缓冲区，而不是底层读者。

Sven Marnach 为常规 Iterators 和

Peekable

 的类似问题提供了出色的 

answer