能否在非Copy类型的异步流中使用futures::StreamExt::take_while？

问题背景 #

在使用futures::stream::take_while适配器时，官方示例大多针对Copy类型的Item。但当流的Item为非Copy类型（比如示例中的Result<Message, Error>），只要在异步断言逻辑中引用Item，就会触发生命周期错误——编译器认为异步块捕获的引用可能在使用前失效。

原示例代码 #

use std::pin::Pin;

use futures_util::{Stream, StreamExt};
use tokio_tungstenite::tungstenite::error::Error;
use tokio_tungstenite::tungstenite::protocol::Message;


async fn process_messages(mut read: Pin<&mut impl Stream<Item = Result<Message, Error>>>) {
    read
     .take_while(|x: &Result<Message, tungstenite::Error>| 
         async { x; true } // 移除`x;`即可编译，但无法使用Item
     );
}

#[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);

    process_messages(read).await;
}

错误信息 #

error: lifetime may not live long enough
--> src/main.rs:13:63
|
13 | read.take_while(|x: &Result<Message, tungstenite::Error>| async { x; true });
| - - ^^^^^^^^^^^^^^^^^ returning this value requires that '1 must outlive '2
| | |
| | return type of closure [async block@src/main.rs:13:63: 13:80] contains a lifetime '2
| let's call the lifetime of this reference '1

error[E0373]: async block may outlive the current function, but it borrows x, which is owned by the current function
--> src/main.rs:13:63
|
13 | read.take_while(|x: &Result<Message, tungstenite::Error>| async { x; true });
| ^^^^^^^^-^^^^^^^
| | |
| | x is borrowed here
| may outlive borrowed value x

解决方案 #

在异步块前添加move关键字，让异步块捕获闭包参数的引用所有权，编译器就能正确推导生命周期关系：

use std::pin::Pin;

use futures_util::{Stream, StreamExt};
use tokio_tungstenite::tungstenite::error::Error;
use tokio_tungstenite::tungstenite::protocol::Message;


async fn process_messages(mut read: Pin<&mut impl Stream<Item = Result<Message, Error>>>) {
    read
     .take_while(|x: &Result<Message, tungstenite::Error>| 
         async move { 
             // 这里可以安全使用x进行判断，比如检查是否为文本消息
             let keep_going = matches!(x, Ok(Message::Text(_)));
             keep_going
         }
     )
     .for_each(|msg| async {
         // 处理符合条件的消息
         println!("处理消息: {:?}", msg);
     })
     .await;
}

#[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);

    process_messages(read).await;
}

原理说明 #

• 默认情况下，异步块会借用闭包参数x，但异步块的执行周期独立于闭包调用周期，编译器无法保证引用在异步块执行时依然有效。
• 使用async move后，异步块会获取x的引用所有权（注意：仅转移引用的所有权，不会消耗原流中的Item），编译器可以明确生命周期绑定，确保引用在异步块执行期间始终有效。
• 由于take_while的闭包参数本身就是对Item的共享引用，move操作完全符合流处理的语义，不会影响后续流适配器对Item的使用。

内容的提问来源于stack exchange，提问作者jsstuball