为hyper::body::Sender实现AsyncWrite时遭遇生命周期错误

要实现hyper::body::Sender到tokio::io::AsyncWrite的转换，核心问题在于Sender::send_data返回的future持有Sender的可变引用，无法满足'static生命周期要求。下面是修正后的完整实现，同时修复了原代码中poll_shutdown不符合AsyncWrite语义的问题：

use hyper::{Request, Body, body::Sender, Response, Error as HyperError}; // 0.14.26
use futures::{Future, FutureExt}; // 0.3.28
use std::task::Poll;
use pin_project::pin_project; // 1.1.0
use tokio::io::{AsyncWrite, Error as IoError, ErrorKind}; // 1.28.2
use bytes::Bytes; // 1.4.0
use std::pin::Pin;

#[pin_project]
pub struct SenderWriter {
    sender: Sender,
    #[pin]
    write_fut: Option<Pin<Box<dyn Future<Output = Result<(), HyperError>> + Send>>>,
    last_len: usize,
    is_closed: bool,
}

impl SenderWriter {
    pub fn new(sender: Sender) -> SenderWriter {
        SenderWriter {
            sender,
            write_fut: None,
            last_len: 0,
            is_closed: false,
        }
    }
}

impl AsyncWrite for SenderWriter {
    fn poll_write(
        self: Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
        buf: &[u8],
    ) -> Poll<Result<usize, IoError>> {
        let mut this = self.project();

        // 已关闭则拒绝写入
        if *this.is_closed {
            return Poll::Ready(Err(IoError::new(ErrorKind::BrokenPipe, "Writer is closed")));
        }

        if this.write_fut.is_none() {
            *this.last_len = buf.len();
            // 将send_data的future包装为动态类型，自动处理生命周期
            let fut = this.sender.send_data(Bytes::copy_from_slice(buf)).boxed();
            *this.write_fut = Some(fut);
        }

        let last_len = *this.last_len;
        let polled = this.write_fut.as_mut().as_pin_mut().unwrap().poll(cx);

        if polled.is_ready() {
            *this.last_len = 0;
            *this.write_fut = None;
        }

        polled.map(|res| res.map(|_| last_len).map_err(|e| IoError::new(ErrorKind::Other, e)))
    }

    fn poll_flush(
        self: Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
    ) -> Poll<Result<(), IoError>> {
        let this = self.project();
        // 等待Sender有空间接收新数据，确保之前的数据已被处理
        this.sender.poll_ready(cx).map(|r| r.map_err(|e| IoError::new(ErrorKind::Other, e)))
    }

    fn poll_shutdown(
        mut self: Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
    ) -> Poll<Result<(), IoError>> {
        let mut this = self.project();

        if *this.is_closed {
            return Poll::Ready(Ok(()));
        }

        // 先确保所有待写入数据完成
        match self.as_mut().poll_flush(cx) {
            Poll::Pending => return Poll::Pending,
            Poll::Ready(Err(e)) => return Poll::Ready(Err(e)),
            Poll::Ready(Ok(())) => {}
        }

        // 启动关闭流程
        if this.write_fut.is_none() {
            let fut = this.sender.close().boxed();
            *this.write_fut = Some(fut);
        }

        let polled = this.write_fut.as_mut().as_pin_mut().unwrap().poll(cx);

        if polled.is_ready() {
            *this.is_closed = true;
            *this.write_fut = None;
        }

        polled.map(|res| res.map_err(|e| IoError::new(ErrorKind::Other, e)))
    }
}

pub async fn my_reusable_fn<W: AsyncWrite + Send + Unpin + 'static>(_writer: W) {
    // 你的通用IO逻辑
}

pub async fn download_handler(_req: Request<Body>) -> Response<Body> {
    let (sender, body) = Body::channel();
    let sender_writer = SenderWriter::new(sender);
    tokio::spawn(my_reusable_fn(sender_writer));
    Response::builder().body(body).unwrap()
}

关键修正点： #

• 移除静态生命周期限制：将BoxFuture<'static>替换为Pin<Box<dyn Future<Output = Result<(), HyperError>> + Send>>，使用FutureExt::boxed()自动处理send_data返回future的生命周期，避免因持有Sender引用导致的生命周期不匹配。
• 完善关闭逻辑：新增is_closed状态标记，在poll_shutdown中先完成所有待写入数据，再调用Sender::close()并等待关闭完成，符合AsyncWrite的语义要求。
• 写入保护：在poll_write中检查is_closed状态，避免已关闭的Writer被重复写入。

内容的提问来源于stack exchange，提问作者Ramesh Kithsiri HettiArachchi