rust tokio trait bounds were not satisfied on forward method

Question

I upgraded wrap and tokio in my rust project and after the upgrade, the forward method got an error. I searched through the documentation but there was no forward method in the new version of tokio framework.

Error

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`

Code:

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 Dependecies:

[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

The most telling lines of that error message are the following:

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

The forward method is defined in the StreamExt trait; due to a blanket implementation, anything that implements Stream also implements StreamExt. However, as of Tokio v1.6.0, UnboundedReceiver no longer implements Stream. The documentation instead states:

This receiver can be turned into a Stream using UnboundedReceiverStream.

Hence:

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);
    }
}));