如何在安全Rust中实现FutureSource转Stream适配器？

问题背景 #

有一个语义上表现为流但未实现futures_util::Stream trait的类型FutureSource，它通过get_next方法反复返回产出下一项的Future。尝试为其实现适配器以当作Stream使用时，遇到了编译错误。

示例代码 #

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

struct FutureSource {/* omitted */}

impl FutureSource {
    pub fn get_next(&mut self) -> FutureString {
        FutureString {
            source: self,
            important_state: todo!(),
        }
    }
}

struct FutureString<'a> {
    source: &'a mut FutureSource,
    important_state: String,
}

impl<'a> Future for FutureString<'a> {
    type Output = String;

    fn poll(self: Pin<&mut Self>, _cx: &mut std::task::Context<'_>) -> Poll<Self::Output> {
        unimplemented!()
    }
}

struct StreamAdapter<'a> {
    future_source: &'a mut FutureSource,
    future: Option<FutureString<'a>>,
}

impl<'a> Stream for StreamAdapter<'a> {
    type Item = <FutureString<'a> as Future>::Output;

    fn poll_next(
        mut self: Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
    ) -> Poll<Option<Self::Item>> {
        if self.future.is_none() {
            self.future.insert(self.future_source.get_next());
        }
        let fut = self.future.as_mut().unwrap();

        match Pin::new(fut).poll(cx) {
            Poll::Ready(value) => {
                self.future = None;
                Poll::Ready(Some(value))
            }
            Poll::Pending => Poll::Pending,
        }
    }
}

编译错误信息 #

error[E0499]: cannot borrow `self` as mutable more than once at a time
  --> src/main.rs:41:32
   |
41 |             self.future.insert(self.future_source.get_next());
             ----        ------ ^^^^ second mutable borrow occurs here
             |           |
             |           first borrow later used by call
             first mutable borrow occurs here

error: lifetime may not live long enough
  --> src/main.rs:41:32
   |
33 | impl<'a> Stream for StreamAdapter<'a> {
      -- lifetime `'a` defined here
...
37 |         mut self: Pin<&mut Self>,
                       - let's call the lifetime of this reference `'1`
...
41 |             self.future.insert(self.future_source.get_next());
                                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ argument requires that `'1` must outlive `'a`

error[E0499]: cannot borrow `self` as mutable more than once at a time
  --> src/main.rs:43:19
   |
33 | impl<'a> Stream for StreamAdapter<'a> {
      -- lifetime `'a` defined here
...
41 |             self.future.insert(self.future_source.get_next());
                                -----------------------------
                                |
                                first mutable borrow occurs here
                                argument requires that `self` is borrowed for `'a`
42 |         }
43 |         let fut = self.future.as_mut().unwrap();
                   ^^^^ second mutable borrow occurs here

error[E0499]: cannot borrow `self` as mutable more than once at a time
  --> src/main.rs:47:17
   |
33 | impl<'a> Stream for StreamAdapter<'a> {
      -- lifetime `'a` defined here
...
41 |             self.future.insert(self.future_source.get_next());
                                -----------------------------
                                |
                                first mutable borrow occurs here
                                argument requires that `self` is borrowed for `'a`
...
47 |                 self.future = None;
                 ^^^^ second mutable borrow occurs here

错误原因分析 #

• 可变借用冲突：self.future_source.get_next()返回的FutureString持有&'a mut FutureSource，该借用的生命周期与StreamAdapter的'a绑定，相当于独占了future_source。此时修改self.future（比如insert或设为None）会尝试再次可变借用self，触发Rust的可变借用独占规则。
• 生命周期不匹配：Stream的poll_next方法接收的self引用生命周期'1无法被编译器证明足够覆盖'a，因为FutureString需要持有'a生命周期的引用，而poll_next的调用周期是不确定的。

安全Rust解决方案 #

不需要使用unsafe代码，核心是调整FutureSource的状态管理方式，避免独占式可变借用：

方案：使用内部可变性分离引用权限 #

将FutureSource的内部状态用RefCell（单线程场景）或Mutex/RwLock（多线程异步场景）包裹，让get_next返回的Future持有共享引用而非独占可变引用，从而消除借用冲突。

修改后的代码示例：

use futures_util::Stream;
use std::{cell::RefCell, future::Future, pin::Pin, task::Poll};

// 定义FutureSource的内部状态（示例）
#[derive(Default)]
struct SourceState {
    count: u32,
}

struct FutureSource {
    state: RefCell<SourceState>, // 使用RefCell实现单线程内部可变性
}

impl FutureSource {
    pub fn get_next(&self) -> FutureString {
        FutureString {
            source: self,
            important_state: "example".to_string(),
        }
    }
}

struct FutureString<'a> {
    source: &'a FutureSource, // 改为共享引用
    important_state: String,
}

impl<'a> Future for FutureString<'a> {
    type Output = String;

    fn poll(self: Pin<&mut Self>, _cx: &mut std::task::Context<'_>) -> Poll<Self::Output> {
        // 通过RefCell获取可变权限修改内部状态
        let mut state = self.source.state.borrow_mut();
        state.count += 1;
        Poll::Ready(format!("Item {}", state.count))
    }
}

struct StreamAdapter<'a> {
    future_source: &'a FutureSource,
    future: Option<FutureString<'a>>,
}

impl<'a> Stream for StreamAdapter<'a> {
    type Item = String;

    fn poll_next(
        mut self: Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
    ) -> Poll<Option<Self::Item>> {
        if self.future.is_none() {
            self.future.insert(self.future_source.get_next());
        }
        let fut = self.future.as_mut().unwrap();

        match Pin::new(fut).poll(cx) {
            Poll::Ready(value) => {
                self.future = None;
                Poll::Ready(Some(value))
            }
            Poll::Pending => Poll::Pending,
        }
    }
}

方案说明 #

• 内部可变性：RefCell允许在共享引用下修改内部状态，Rust会在运行时检查借用规则，确保同一时间只有一个可变借用。
• 消除独占借用：FutureString持有&FutureSource而非&mut FutureSource，不会独占整个源，因此StreamAdapter可以自由修改future字段，避免了借用冲突。
• 多线程适配：如果需要在多线程异步环境中使用，只需将RefCell替换为tokio::sync::Mutex或async_std::sync::Mutex等异步安全的锁结构。

结论 #

你的需求是健全的，完全可以在安全Rust中实现该适配器，不需要依赖unsafe代码。核心是通过内部可变性调整引用权限，规避Rust的借用检查限制。

内容的提问来源于stack exchange，提问作者Emil Sahlén