rust tokio trait bound 在 forward 方法上不满足

Question

我升级了 wrap和 tokio在我的 Rust 项目和升级后，forward 方法出错了。我搜索了文档，但在新版本的 tokio 框架中没有转发方法。
错误

error[E0599]: the method `forward` exists for struct `tokio::sync::mpsc::UnboundedReceiver<_>`, but its trait bounds were not satisfied


tokio::task::spawn(client_rcv.forward(client_ws_sender).map(|result| {
                                      ^^^^^^^ method cannot be called on `tokio::sync::mpsc::UnboundedReceiver<_>` due to unsatisfied trait bounds
40 | pub struct UnboundedReceiver<T> {
   | -------------------------------
   | |
   | doesn't satisfy `_: warp::Stream`
   | doesn't satisfy `tokio::sync::mpsc::UnboundedReceiver<_>: StreamExt`
   |
   = note: the following trait bounds were not satisfied:
           `tokio::sync::mpsc::UnboundedReceiver<_>: warp::Stream`
           which is required by `tokio::sync::mpsc::UnboundedReceiver<_>: StreamExt`
           `&tokio::sync::mpsc::UnboundedReceiver<_>: warp::Stream`
           which is required by `&tokio::sync::mpsc::UnboundedReceiver<_>: StreamExt`
           `&mut tokio::sync::mpsc::UnboundedReceiver<_>: warp::Stream`
           which is required by `&mut tokio::sync::mpsc::UnboundedReceiver<_>: StreamExt`

代码:

let (client_ws_sender, mut client_ws_rcv) = ws.split();
let (client_sender, client_rcv) = mpsc::unbounded_channel();

tokio::task::spawn(client_rcv.forward(client_ws_sender).map(|result| {
    if let Err(e) = result {
        eprintln!("error sending websocket msg: {}", e);
    }
}));

cargo 依赖:

[dependencies]
tokio = { version = "1.6.0", features = ["full"] }
warp = "0.3.1"
serde = {version = "1.0", features = ["derive"] }
serde_json = "1.0"
futures = { version = "0.3", default-features = false }
uuid = { version = "0.8.2", features = ["serde", "v4"] }

Answer 1

该错误消息中最有说服力的行如下:

   | doesn't satisfy `_: warp::Stream`
   | doesn't satisfy `tokio::sync::mpsc::UnboundedReceiver<_>: StreamExt`

forward 方法在 StreamExt 中定义特征;由于 a blanket implementation , 任何实现 Stream 的东西还实现了 StreamExt .但是，从 Tokio v1.6.0 开始， UnboundedReceiver 不再实现 Stream .文档反而指出:

This receiver can be turned into a Stream using UnboundedReceiverStream.

因此:

let (client_ws_sender, mut client_ws_rcv) = ws.split();
let (client_sender, client_rcv) = mpsc::unbounded_channel();
let client_rcv = UnboundedReceiverStream::new(client_rcv);  // <-- this

tokio::task::spawn(client_rcv.forward(client_ws_sender).map(|result| {
    if let Err(e) = result {
        eprintln!("error sending websocket msg: {}", e);
    }
}));