Rust: tungstenite + warp doesn't send

Question

Consider this minimal example of a server-client websocket in Rust.

Server:

use std::collections::VecDeque;
use rand::prelude::*;
use rand::Rng; // 0.8.5
use rand_distr::StandardNormal;
use rudy::rudymap::RudyMap;
mod matcher;
use matcher::EngineOrderedTree;
use futures_util::{FutureExt, StreamExt};
use warp::Filter;
use futures_util::SinkExt;

use std::{convert::Infallible, net::SocketAddr};
use hyper::{Body, Request, Response, Server};
use hyper::service::{make_service_fn, service_fn};
use std::sync::Arc;

#[tokio::main]
async fn main() {
    let routes = warp::path("echo")
        // The `ws()` filter will prepare the Websocket handshake.
        .and(warp::ws())
        .map(|ws: warp::ws::Ws| {
            // And then our closure will be called when it completes...
            ws.on_upgrade(|websocket| {
                // Just echo all messages back...
                let (tx, rx) = websocket.split();
                rx.for_each(|message| async {
                    match message {
                        Err(e) => {
                            eprintln!("websocket error: {:?}", e);
                        },
                        Ok(message) => {
                            println!("{:?}", message);
                        },
                    }
                })
            })
        });

    warp::serve(routes).run(([127, 0, 0, 1], 8000)).await;
}

Client:

use tokio_tungstenite::{connect_async, tungstenite::protocol::Message};
use url::Url;
use chrono::Utc;
use tokio::sync::Mutex;
use futures_util::{future, pin_mut, StreamExt};
use std::sync::Arc;
use std::env;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use futures_util::SinkExt;

#[tokio::main]
async fn main() {
    let url = url::Url::parse("ws://localhost:8000").unwrap();

    let (ws_stream, _) = connect_async(url).await.expect("Failed to connect");
    println!("WebSocket handshake has been successfully completed");

    let (mut write, read) = ws_stream.split();

    let mut i: Arc> = Arc::new(Mutex::new(0));
    let write_to_ws = async {
        for i in 1.. {
            write.send(Message::from("a"));
            write.flush();
        }
    };

    let read_ws = read.for_each(|message| async {
        println!("a");
        let data = message.unwrap().into_data();
        let mut m = i.lock().await;
        *m += 1;
        if *m % 1 == 0 {
            println!("{:?}", m);
        }
    });

    pin_mut!(write_to_ws, read_ws);
    future::select(write_to_ws, read_ws).await;
}

The project compiles perfectly well and everything appears as it would if it were working. But it doesn't. The messages aren't getting echoed back from the server.

How do you even debug this? There are no compilation errors to solve and calling send and read.for_each seem to be the the functions that are in charge of sending/receiving data.

Rust: tungstenite + warp doesn't send

0 Answers