包装fastwebsockets FragmentCollector实现Stream的Pinning问题求助

问题背景 #

你尝试将fastwebsockets::FragmentCollector包装成实现Stream<Item = Vec<u8>>的结构体，当前的结构体定义及Stream实现代码如下：

结构体定义 #

struct MessageStream {
    inner: FragmentCollector<TokioIo<Upgraded>>,
    fut: Option<BoxFuture<'static, Result<Vec<u8>, Whatever>>>,
}

待修正的poll_next实现 #

fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Result<Vec<u8>, Whatever>>> {
    match self.fut {
        Some(fut) => match fut.poll(cx) { // Error: Pin<Box<dyn Future<...>>> doesn't implement Future because the future isn't Unpin
            Poll::Pending => {
                self.fut.replace(fut);
                Poll::Pending
            }
            Poll::Ready(Err(e)) => Poll::Ready(Some(Err(e))),
            Poll::Ready(Ok(data)) => Poll::Ready(Some(Ok(data)))
        }
        None => {
            self.fut = Some(Box::pin(self.read_next()));
            self.poll_next(cx)
        }
    }
}

read_next方法 #

async fn read_next(&mut self) -> Result<Vec<u8>, Whatever> {
    loop {
        match self.inner.read_frame().await {
            Err(e) => whatever!("failed to read websocket frame: {}", e),
            Ok(frame) => match frame.opcode {
                OpCode::Close => whatever!("websocket closed"),
                OpCode::Text | OpCode::Binary => match frame.payload {
                    fastwebsockets::Payload::Owned(data) => return Ok(data),
                    _ => unreachable!(), // Depends on implementation detail of FragmentCollector, but I'm ok with that.
                },
                _ => trace!("ignoring frame with opcode {:?}", frame.opcode),
            },
        }
    }
}

错误原因分析 #

1. 生命周期不匹配：read_next捕获了&mut self，生成的future生命周期与MessageStream实例绑定，但你给BoxFuture指定了'static生命周期，这会导致编译错误——future无法脱离self独立存在。
2. Pin使用错误：BoxFuture本质是Pin<Box<dyn Future<...>>>，要轮询它必须获取Pin<&mut dyn Future<...>>，直接调用fut.poll(cx)不符合Pin的规则，因为fut是被Pin在Box里的，必须通过Pin的引用去poll。

解决方案 #

方案1：修正手动Stream实现 #

步骤1：调整结构体生命周期

把BoxFuture的'static改为与结构体实例绑定的生命周期：

use std::pin::Pin;
use futures::{Future, Stream, StreamExt};
use futures::future::BoxFuture;
use std::task::{Context, Poll};

struct MessageStream {
    inner: FragmentCollector<TokioIo<Upgraded>>,
    fut: Option<BoxFuture<'_, Result<Vec<u8>, Whatever>>>, // 使用'_绑定结构体生命周期
}

步骤2：正确实现poll_next

在处理fut时，要正确获取Pin引用，并且避免所有权转移导致的问题：

impl Stream for MessageStream {
    type Item = Result<Vec<u8>, Whatever>;

    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        // 先处理已存在的future
        if let Some(mut fut) = self.fut.take() {
            match Pin::as_mut(&mut fut).poll(cx) {
                Poll::Pending => {
                    // 把未完成的future放回
                    self.fut = Some(fut);
                    Poll::Pending
                }
                Poll::Ready(Err(e)) => Poll::Ready(Some(Err(e))),
                Poll::Ready(Ok(data)) => Poll::Ready(Some(Ok(data))),
            }
        } else {
            // 创建新future并立即轮询
            let fut = Box::pin(self.read_next());
            self.fut = Some(fut);
            self.poll_next(cx)
        }
    }
}

方案2：使用async_stream宏简化实现（推荐） #

手动实现Stream容易出错，Tokio生态中的async_stream宏可以让你用async/await语法直接生成Stream，无需处理Pin和poll逻辑：

首先添加依赖到Cargo.toml：

[dependencies]
async-stream = "0.3"
futures = "0.3"

然后重写MessageStream：

use async_stream::stream;
use futures::Stream;

struct MessageStream {
    inner: FragmentCollector<TokioIo<Upgraded>>,
}

impl MessageStream {
    fn into_stream(self) -> impl Stream<Item = Result<Vec<u8>, Whatever>> {
        stream! {
            let mut inner = self.inner;
            loop {
                match inner.read_frame().await {
                    Err(e) => yield Err(whatever!("failed to read websocket frame: {}", e)),
                    Ok(frame) => match frame.opcode {
                        OpCode::Close => {
                            yield Err(whatever!("websocket closed"));
                            break;
                        }
                        OpCode::Text | OpCode::Binary => match frame.payload {
                            fastwebsockets::Payload::Owned(data) => yield Ok(data),
                            _ => unreachable!(),
                        },
                        _ => trace!("ignoring frame with opcode {:?}", frame.opcode),
                    },
                }
            }
        }
    }
}

这种方式完全避开了手动处理Pin和poll_next的复杂逻辑，代码更简洁易读。

代码风格与实现思路建议 #

1. 优先使用高层抽象：尽量用async_stream、StreamExt等工具库，避免手动实现Stream/Future，除非你需要极致的性能或特殊逻辑。
2. 避免不必要的'static：生命周期标注要精准，'static只有在future需要脱离原所有者存在时才使用，大多数场景下用与实例绑定的生命周期即可。
3. Pin的使用原则：如果一个future包含对可变引用的捕获，它通常不满足Unpin，必须用Pin来固定；BoxFuture已经是Pin过的类型，操作时要通过Pin::as_mut获取可变Pin引用。
4. 错误处理优化：whatever!宏如果是自定义的错误转换，建议明确错误类型，避免模糊的错误信息；对于Close帧，可以选择返回None来结束Stream，而不是返回错误，更符合Stream的语义（Stream结束时返回None）。
5. 减少实现细节依赖：unreachable!()依赖FragmentCollector的实现细节，建议添加防御性代码，比如返回一个错误而不是panic，避免未来库更新导致程序崩溃。

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