Borrowed data escapes outside of method - tokio accept async

I have a Rust function api.ws_kline(...) connecting to a Websocket API. The Rust function receives data from that API in a stream. Using transmitters and receivers I send the transmitter to the API function api.ws_kline(...) so the receiver can retrieve it and pass it back to the client connecting with a websocket to my own API.

So two websockets.

  • One websocket for the client connecting to my app.
  • One websocket for my app to connect to the external API.
    -> Now I want to transmit the data received from the external API to the other websocket.
    pub async fn btc_candlestick(&self, req: http::Http, stream: &mut TcpStream, psql: Psql, api: Api) -> Result<http::HttpResponse, http::Error> {

        // Upgrade connection to websocket.
        let ws_stream = accept_async(stream).await.expect("Failed to accept");
        let (mut ws_sender, mut ws_receiver) = ws_stream.split();

        // Make transmitter and receiver. Transmitter is passed to ws_kline so receiver can
        // send it back to the client requesting.
        let (tx, rx) = mpsc::channel::<CandleStick>(10);
        let tx = Arc::new(Mutex::new(tx));
        let rx = Arc::new(Mutex::new(rx));

        // Get klines from API.
        tokio::spawn(async move {
            api.ws_kline(tx, "1s".into()).await.unwrap_or(());
        });
      
        // Send messages from rx to client of our websocket.
        tokio::spawn(async move {
            while let Some(c) = rx.lock().await.recv().await {    
                // Adding this line generates error.
                ws_sender.send(Message::Text("tick".to_owned())).await.unwrap();
            }
        });
    

        Ok(http::HttpResponse{
            status: 200,
            headers: vec![
                ("Content-Type".into(), "text/plain".into()),
            ],
            body: "".into(),
        })
    }

Error:

error[E0521]: borrowed data escapes outside of method
   --> src/routes/mod.rs:774:25
    |
771 | ...candlestick(&self, req: http::Http, stream: &mut TcpStream, psql: Psql, api: A...
    |                                        ------  -                           --- `api` declared here, outside of the method body
    |                                        |       |
    |                                        |       let's call the lifetime of this reference `'1`
    |                                        `stream` is a reference that is only valid in the method body
...
774 | ... = accept_async(stream).await.expect("Failed to accept");
    |       ^^^^^^^^^^^^^^^^^^^^
    |       |
    |       `stream` escapes the method body here
    |       argument requires that `'1` must outlive `'static`

Just pass the TcpStream by value and not by reference.


This probably has to be into_split() to created owned read and write halves that you can pass to your spawned tasks.


It's unnecessary to wrap your sender and receiver in a mutex here, you should remove these two lines.


Your return type also looks superfluous to me. If you upgrade the incoming connection to a websocket, what is this response for? Who do you want to respond to? The connection has been upgraded from http to a websocket.

TcpStream should really be passed as reference though. It is not possible to do it as value because then my whole code base would crash. Stream has passed several other functions before it comes into btc_candlestick changing all those to accept TcpStream as value would result in ownership errors.

And I think an Arc<Mutex<TcpStream>> also isn't a solution because accept_async doesn't accept Arc/Mutex.