为reqwest::Response实现tokio::io::AsyncRead遇生命周期错误求助

我尝试手动为reqwest::Response实现tokio::io::AsyncRead trait（已知可以用tokio_util::io::StreamReader快速转换，但想手动实现），写了如下代码：

use std::future::Future;
use std::pin::{pin, Pin};
use std::task::Poll;

use bytes::Bytes;
use reqwest::Response;
use reqwest::Result;

use tokio::io::AsyncRead;
use pin_project::pin_project;


#[pin_project]
struct ResponseStreamPin {
    inner: Response,
    future: Option<Pin<Box<dyn Future<Output = Result<Option<Bytes>>>>>>
}

impl AsyncRead for ResponseStreamPin {
    fn poll_read(
        mut self: Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
        buf: &mut tokio::io::ReadBuf<'_>,
    ) -> Poll<std::io::Result<()>> {
        let s = self.as_mut().project();

        if let Some(f) = s.future {
            match f.as_mut().poll(cx) {
                Poll::Ready(result) => {
                    s.future.take();
                    match result {
                        Ok(maybe_bytes) => {
                            if let Some(bytes) = maybe_bytes {
                                buf.put_slice(&bytes);
                            };
                            return Poll::Ready(Ok(()))
                        },
                        Err(_) => return Poll::from(Err(std::io::Error::last_os_error())),
                    }
                },
                _ => return Poll::Pending,
            }
        }

        let future = Box::pin(s.inner.chunk());
        *s.future = Some(future);
        return self.poll_read(cx, buf);
    }
}

但编译时出现以下错误：

error[E0597]: `self` does not live long enough
  --> client/src/async_client/response_stream.rs:26:17
   |
22 |         mut self: Pin<&mut Self>,
   |         -------- binding `self` declared here
...
26 |         let s = self.as_mut().project();
   |                 ^^^^ borrowed value does not live long enough
...
47 |         *s.future = Some(future);
   |                          ------ cast requires that `self` is borrowed for `'static`
48 |         return self.poll_read(cx, buf);
49 |     }
   |      - `self` dropped here while still borrowed

error: lifetime may not live long enough
  --> client/src/async_client/response_stream.rs:47:26
   |
22 |         mut self: Pin<&mut Self>,
   |                       - let's call the lifetime of this reference `'1`
...
47 |         *s.future = Some(future);
   |                          ^^^^^^ cast requires that `'1` must outlive `'static`

error[E0505]: cannot move out of `self` because it is borrowed
  --> client/src/async_client/response_stream.rs:48:16
   |
22 |         mut self: Pin<&mut Self>,
   |         -------- binding `self` declared here
...
26 |         let s = self.as_mut().project();
   |                 ---- borrow of `self` occurs here
...
47 |         *s.future = Some(future);
   |                          ------ cast requires that `self` is borrowed for `'static`
48 |         return self.poll_read(cx, buf);
   |                ^^^^ move out of `self` occurs here

Some errors have detailed explanations: E0505, E0597.
For more information about an error, try `rustc --explain E0505`.

问题分析与解决 #

1. 核心问题：dyn Future的生命周期约束

代码中future字段使用的Pin<Box<dyn Future<...>>>默认要求'static生命周期，但response.chunk()返回的future实际与Response的生命周期绑定，无法满足'static要求，这直接导致了生命周期不匹配的错误。同时，project()返回的引用与后续递归调用self.poll_read产生了借用冲突。

2. 具体修复步骤

（1）为结构体添加生命周期参数，绑定future与内部Response的生命周期

修改ResponseStreamPin定义，引入生命周期'a，让future的生命周期与结构体内部的Response绑定，避免强制要求'static：

#[pin_project]
struct ResponseStreamPin<'a> {
    inner: &'a mut Response,
    future: Option<Pin<Box<dyn Future<Output = Result<Option<Bytes>>> + 'a>>>
}

这里将inner改为可变引用（因为chunk()方法需要&mut self），同时给dyn Future加上'a生命周期约束。

（2）调整poll_read逻辑，解决借用冲突与错误转换问题

重新组织代码逻辑，确保project()返回的引用作用域不会覆盖递归调用，同时正确转换reqwest错误为IO错误：

impl<'a> AsyncRead for ResponseStreamPin<'a> {
    fn poll_read(
        mut self: Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
        buf: &mut tokio::io::ReadBuf<'_>,
    ) -> Poll<std::io::Result<()>> {
        let mut this = self.as_mut().project();

        // 处理正在进行的chunk请求
        if let Some(future) = this.future.as_mut() {
            match future.poll(cx) {
                Poll::Ready(res) => {
                    this.future.take(); // 清空已完成的future
                    match res {
                        Ok(Some(bytes)) => {
                            buf.put_slice(&bytes);
                            Poll::Ready(Ok(()))
                        }
                        Ok(None) => {
                            // 响应流已结束，返回读取完成
                            Poll::Ready(Ok(()))
                        }
                        Err(e) => {
                            // 将reqwest错误转换为IO错误
                            Poll::Ready(Err(std::io::Error::new(
                                std::io::ErrorKind::Other,
                                e.to_string(),
                            )))
                        }
                    }
                }
                Poll::Pending => Poll::Pending,
            }
        } else {
            // 发起新的chunk请求
            let future = Box::pin(this.inner.chunk());
            *this.future = Some(future);
            // 此时this的作用域已结束，递归调用不会触发借用冲突
            self.poll_read(cx, buf)
        }
    }
}

（3）补充结构体构造方法

方便创建ResponseStreamPin实例：

impl<'a> ResponseStreamPin<'a> {
    fn new(inner: &'a mut Response) -> Self {
        Self {
            inner,
            future: None,
        }
    }
}

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