Struct Client

Source

pub struct Client { /* private fields */ }

Expand description

A WebSocket client that connects to a Server (or any compatible WebSocket server) and exchanges events with it.

§Example

use rustwave::{Client, Result};

#[tokio::main]
async fn main() -> Result<()> {
    let client = Client::connect("ws://127.0.0.1:8080").await?;

    client.on("message", |data| async move {
        println!("Server say: {}", data);
    });

    client.emit("message", "hello").await?;
    client.wait().await;

    Ok(())
}

Implementations§

Source §

impl Client

Source

pub async fn connect(url: &str) -> Result<Self>

Connects to a WebSocket server at the given url.

On success, this establishes the connection and spawns background tasks to handle reading incoming events, sending outgoing events, and keeping the connection alive with periodic heartbeat pings. Event handlers can then be registered with Client::on and messages sent with Client::emit.

§Errors

Returns an error if the connection to url cannot be established.

§Example

use rustwave::{Client, Result};

#[tokio::main]
async fn main() -> Result<()> {
    let client = Client::connect("ws://127.0.0.1:8080").await?;

    client.on("message", |data| async move {
        println!("Server say: {}", data);
    });

    Ok(())
}

Examples found in repository ?

examples/test1.rs (line 5)

4async fn main() -> Result<()> {
5    let client = Client::connect("ws://127.0.0.1:8080/").await?;
6
7    client.on("pong", |data| async move {
8        println!("Server replied: {}", data);
9    });
10
11    client.emit("ping", "").await?;
12
13    tokio::time::sleep(std::time::Duration::from_secs(1)).await;
14
15    Ok(())
16}

More examples

Hide additional examples

examples/test2.rs (line 5)

4async fn main() -> Result<()> {
5    let client = Client::connect("ws://127.0.0.1:8080/chat").await?;
6
7    client.on("message", |data| async move {
8        println!("Got back: {}", data);
9    });
10
11    client.emit("mess", "hello from chat client").await?;
12
13    tokio::time::sleep(std::time::Duration::from_secs(1)).await;
14
15    client.wait().await;
16
17    Ok(())
18}

examples/terminal_client.rs (line 14)

13async fn main() -> Result<()> {
14    let client = Client::connect("ws://127.0.0.1:8080").await?;
15
16    client.on("chat", |data| async move {
17        println!("Room message:> {}", data);
18    });
19
20    let room = "chat";
21    client.emit("join-room", room).await?;
22
23    let mut stdin = BufReader::new(tokio::io::stdin());
24
25    loop {
26        print!("you> ");
27        std::io::stdout().flush().unwrap();
28
29        let mut line = String::new();
30        if stdin.read_line(&mut line).await.unwrap() == 0 {
31            break; 
32        }
33
34        let text = line.trim().to_string();
35        if text.is_empty() {
36            continue;
37        }
38
39        let msg = ChatMessage {
40            room: room.to_string(),
41            text,
42        };
43        client.emit("chat", msg).await.unwrap();
44    }
45
46    Ok(())
47}

examples/test3.rs (line 15)

14async fn main() -> Result<()> {
15    let client = Client::connect("ws://127.0.0.1:8080/game").await?;
16
17    client.on("online-count", |data| async move {
18        println!("Online in /game: {}", data);
19    });
20
21    client.on("player-joined", |data| async move {
22        println!("Someone joined the room: {}", data);
23    });
24
25    client.on("player-left", |data| async move {
26        println!("Someone left the room: {}", data);
27    });
28
29    client.on("typing", |data| async move {
30        println!("{}", data);
31    });
32
33    client.on("mess", |data| async move {
34        println!("Room message: {}", data);
35    });
36
37    client.emit("online-count", "").await?;
38    sleep(Duration::from_millis(300)).await;
39
40    client.emit("join-room", ROOM).await?;
41    sleep(Duration::from_millis(300)).await;
42
43    client.emit("typing", ROOM).await?;
44    sleep(Duration::from_millis(300)).await;
45
46    let msg = GameMessage {
47        mess: "hey room, I'm here!".to_string(),
48        room: ROOM.to_string(),
49    };
50    client.emit("mess", msg).await?;
51    sleep(Duration::from_millis(300)).await;
52
53    client.emit("leave-room", ROOM).await?;
54
55    sleep(Duration::from_secs(3)).await;
56
57    Ok(())
58}

examples/client_banch.rs (line 43)

13async fn main() {
14    let addr = "127.0.0.1:9500";
15
16    let client_sent = Arc::new(AtomicU64::new(0));
17    let client_received = Arc::new(AtomicU64::new(0));
18    let client_connect_failed = Arc::new(AtomicU64::new(0));
19
20    let connect_gate = Arc::new(Semaphore::new(MAX_CONCURRENT_CONNECTS));
21
22    println!(
23        "Starting client benchmark: {} clients x {} messages = {} total messages expected",
24        NUM_CLIENTS,
25        MESSAGES_PER_CLIENT,
26        NUM_CLIENTS * MESSAGES_PER_CLIENT
27    );
28
29    let start = Instant::now();
30
31    let mut handles = Vec::with_capacity(NUM_CLIENTS);
32
33    for _ in 0..NUM_CLIENTS {
34        let addr = addr.to_string();
35        let client_sent = client_sent.clone();
36        let client_received = client_received.clone();
37        let client_connect_failed = client_connect_failed.clone();
38        let connect_gate = connect_gate.clone();
39
40        let handle = tokio::spawn(async move {
41            let permit = connect_gate.acquire_owned().await.unwrap();
42
43            let client = match Client::connect(&format!("ws://{}", addr)).await {
44                Ok(c) => c,
45                Err(_) => {
46                    client_connect_failed.fetch_add(1, Ordering::Relaxed);
47                    return None;
48                }
49            };
50
51            drop(permit);
52
53            {
54                let client_received = client_received.clone();
55                client.on("pong", move |_data| {
56                    let client_received = client_received.clone();
57                    async move {
58                        client_received.fetch_add(1, Ordering::Relaxed);
59                    }
60                });
61            }
62
63            for i in 0..MESSAGES_PER_CLIENT {
64                if client.emit("ping", format!("msg-{}", i)).await.is_ok() {
65                    client_sent.fetch_add(1, Ordering::Relaxed);
66                }
67            }
68
69            Some(client)
70        });
71
72        handles.push(handle);
73    }
74
75    let mut clients_kept_alive = Vec::with_capacity(NUM_CLIENTS);
76    for handle in handles {
77        if let Ok(Some(client)) = handle.await {
78            clients_kept_alive.push(client);
79        }
80    }
81
82    let sending_done_at = start.elapsed();
83
84    let expected_total = (NUM_CLIENTS * MESSAGES_PER_CLIENT) as u64;
85
86    let wait_timeout = Duration::from_secs(30);
87    let wait_start = Instant::now();
88
89    loop {
90        let received_now = client_received.load(Ordering::Relaxed);
91        if received_now >= expected_total || wait_start.elapsed() > wait_timeout {
92            break;
93        }
94        sleep(Duration::from_millis(50)).await;
95    }
96
97    let total_elapsed = start.elapsed();
98
99    println!("========== CLIENT-SIDE STATS ==========");
100    println!("Clients:                    {}", NUM_CLIENTS);
101    println!(
102        "Connect failures:           {}",
103        client_connect_failed.load(Ordering::Relaxed)
104    );
105    println!("Messages per client:        {}", MESSAGES_PER_CLIENT);
106    println!("Expected total messages:    {}", expected_total);
107    println!("------------------------------------------");
108    println!(
109        "Sent messages:               {}",
110        client_sent.load(Ordering::Relaxed)
111    );
112    println!(
113        "Replies received:            {}",
114        client_received.load(Ordering::Relaxed)
115    );
116    println!("------------------------------------------");
117    println!("Time to finish sending:     {:?}", sending_done_at);
118    println!("Total time (incl. replies): {:?}", total_elapsed);
119    println!(
120        "Client-side throughput:     {:.0} msgs/sec",
121        expected_total as f64 / total_elapsed.as_secs_f64()
122    );
123    println!("==========================================");
124}

examples/bench.rs (line 68)

13async fn main() {
14    let addr = "127.0.0.1:9500";
15    let server = Server::bind(addr);
16
17    let server_received = Arc::new(AtomicU64::new(0));
18    let server_replied = Arc::new(AtomicU64::new(0));
19
20    {
21        let server_received = server_received.clone();
22        let server_replied = server_replied.clone();
23
24        server.on("ping", move |data, ctx| {
25            let server_received = server_received.clone();
26            let server_replied = server_replied.clone();
27            async move {
28                server_received.fetch_add(1, Ordering::Relaxed);
29                let text = data.parse::<String>().unwrap_or_default();
30                if ctx.emit("pong", text).await.is_ok() {
31                    server_replied.fetch_add(1, Ordering::Relaxed);
32                }
33            }
34        });
35    }
36
37    tokio::spawn(async move {
38        let _ = server.run().await;
39    });
40
41    sleep(Duration::from_millis(300)).await;
42
43    let client_sent = Arc::new(AtomicU64::new(0));
44    let client_received = Arc::new(AtomicU64::new(0));
45
46    let connect_gate = Arc::new(Semaphore::new(MAX_CONCURRENT_CONNECTS));
47
48    println!(
49        "Starting benchmark: {} clients x {} messages = {} total messages expected",
50        NUM_CLIENTS,
51        MESSAGES_PER_CLIENT,
52        NUM_CLIENTS * MESSAGES_PER_CLIENT
53    );
54
55    let start = Instant::now();
56
57    let mut handles = Vec::with_capacity(NUM_CLIENTS);
58
59    for _ in 0..NUM_CLIENTS {
60        let addr = addr.to_string();
61        let client_sent = client_sent.clone();
62        let client_received = client_received.clone();
63        let connect_gate = connect_gate.clone();
64
65        let handle = tokio::spawn(async move {
66            let permit = connect_gate.acquire_owned().await.unwrap();
67
68            let client = match Client::connect(&format!("ws://{}", addr)).await {
69                Ok(c) => c,
70                Err(_) => return None,
71            };
72
73            drop(permit);
74
75            {
76                let client_received = client_received.clone();
77                client.on("pong", move |_data| {
78                    let client_received = client_received.clone();
79                    async move {
80                        client_received.fetch_add(1, Ordering::Relaxed);
81                    }
82                });
83            }
84
85            for i in 0..MESSAGES_PER_CLIENT {
86                if client.emit("ping", format!("msg-{}", i)).await.is_ok() {
87                    client_sent.fetch_add(1, Ordering::Relaxed);
88                }
89            }
90
91            Some(client)
92        });
93
94        handles.push(handle);
95    }
96
97    let mut clients_kept_alive = Vec::with_capacity(NUM_CLIENTS);
98    for handle in handles {
99        if let Ok(Some(client)) = handle.await {
100            clients_kept_alive.push(client);
101        }
102    }
103
104    let sending_done_at = start.elapsed();
105
106    let expected_total = (NUM_CLIENTS * MESSAGES_PER_CLIENT) as u64;
107
108    let wait_timeout = Duration::from_secs(30);
109    let wait_start = Instant::now();
110
111    loop {
112        let received_now = client_received.load(Ordering::Relaxed);
113        if received_now >= expected_total || wait_start.elapsed() > wait_timeout {
114            break;
115        }
116        sleep(Duration::from_millis(50)).await;
117    }
118
119    let total_elapsed = start.elapsed();
120
121    println!();
122    println!("========== BENCHMARK RESULTS ==========");
123    println!("Clients spawned:            {}", NUM_CLIENTS);
124    println!("Messages per client:        {}", MESSAGES_PER_CLIENT);
125    println!("Expected total messages:    {}", expected_total);
126    println!("----------------------------------------");
127    println!(
128        "Client -> sent messages:    {}",
129        client_sent.load(Ordering::Relaxed)
130    );
131    println!(
132        "Server -> received:         {}",
133        server_received.load(Ordering::Relaxed)
134    );
135    println!(
136        "Server -> replies sent:     {}",
137        server_replied.load(Ordering::Relaxed)
138    );
139    println!(
140        "Client -> replies received: {}",
141        client_received.load(Ordering::Relaxed)
142    );
143    println!("----------------------------------------");
144    println!("Time to finish sending:     {:?}", sending_done_at);
145    println!("Total time (incl. replies): {:?}", total_elapsed);
146    println!(
147        "Throughput:                 {:.0} msgs/sec",
148        expected_total as f64 / total_elapsed.as_secs_f64()
149    );
150    println!("=========================================");
151}

Source

pub fn on<F, Fut>(&self, event: &str, callback: F)
where F: Fn(Payload) -> Fut + Send + Sync + 'static, Fut: Future<Output = ()> + Send + 'static,

Registers an event handler on this client.

Registering a handler for an event that already has one replaces the previous handler.

§Example

use rustwave::{Client, Result};

    let client = Client::connect("ws://127.0.0.1:8080").await?;

    client.on("message", |data| async move {
        println!("Server say: {}", data);
    });
    Ok(())

Examples found in repository ?

examples/test1.rs (lines 7-9)

4async fn main() -> Result<()> {
5    let client = Client::connect("ws://127.0.0.1:8080/").await?;
6
7    client.on("pong", |data| async move {
8        println!("Server replied: {}", data);
9    });
10
11    client.emit("ping", "").await?;
12
13    tokio::time::sleep(std::time::Duration::from_secs(1)).await;
14
15    Ok(())
16}

More examples

Hide additional examples

examples/test2.rs (lines 7-9)

4async fn main() -> Result<()> {
5    let client = Client::connect("ws://127.0.0.1:8080/chat").await?;
6
7    client.on("message", |data| async move {
8        println!("Got back: {}", data);
9    });
10
11    client.emit("mess", "hello from chat client").await?;
12
13    tokio::time::sleep(std::time::Duration::from_secs(1)).await;
14
15    client.wait().await;
16
17    Ok(())
18}

examples/terminal_client.rs (lines 16-18)

13async fn main() -> Result<()> {
14    let client = Client::connect("ws://127.0.0.1:8080").await?;
15
16    client.on("chat", |data| async move {
17        println!("Room message:> {}", data);
18    });
19
20    let room = "chat";
21    client.emit("join-room", room).await?;
22
23    let mut stdin = BufReader::new(tokio::io::stdin());
24
25    loop {
26        print!("you> ");
27        std::io::stdout().flush().unwrap();
28
29        let mut line = String::new();
30        if stdin.read_line(&mut line).await.unwrap() == 0 {
31            break; 
32        }
33
34        let text = line.trim().to_string();
35        if text.is_empty() {
36            continue;
37        }
38
39        let msg = ChatMessage {
40            room: room.to_string(),
41            text,
42        };
43        client.emit("chat", msg).await.unwrap();
44    }
45
46    Ok(())
47}

examples/test3.rs (lines 17-19)

14async fn main() -> Result<()> {
15    let client = Client::connect("ws://127.0.0.1:8080/game").await?;
16
17    client.on("online-count", |data| async move {
18        println!("Online in /game: {}", data);
19    });
20
21    client.on("player-joined", |data| async move {
22        println!("Someone joined the room: {}", data);
23    });
24
25    client.on("player-left", |data| async move {
26        println!("Someone left the room: {}", data);
27    });
28
29    client.on("typing", |data| async move {
30        println!("{}", data);
31    });
32
33    client.on("mess", |data| async move {
34        println!("Room message: {}", data);
35    });
36
37    client.emit("online-count", "").await?;
38    sleep(Duration::from_millis(300)).await;
39
40    client.emit("join-room", ROOM).await?;
41    sleep(Duration::from_millis(300)).await;
42
43    client.emit("typing", ROOM).await?;
44    sleep(Duration::from_millis(300)).await;
45
46    let msg = GameMessage {
47        mess: "hey room, I'm here!".to_string(),
48        room: ROOM.to_string(),
49    };
50    client.emit("mess", msg).await?;
51    sleep(Duration::from_millis(300)).await;
52
53    client.emit("leave-room", ROOM).await?;
54
55    sleep(Duration::from_secs(3)).await;
56
57    Ok(())
58}

examples/client_banch.rs (lines 55-60)

13async fn main() {
14    let addr = "127.0.0.1:9500";
15
16    let client_sent = Arc::new(AtomicU64::new(0));
17    let client_received = Arc::new(AtomicU64::new(0));
18    let client_connect_failed = Arc::new(AtomicU64::new(0));
19
20    let connect_gate = Arc::new(Semaphore::new(MAX_CONCURRENT_CONNECTS));
21
22    println!(
23        "Starting client benchmark: {} clients x {} messages = {} total messages expected",
24        NUM_CLIENTS,
25        MESSAGES_PER_CLIENT,
26        NUM_CLIENTS * MESSAGES_PER_CLIENT
27    );
28
29    let start = Instant::now();
30
31    let mut handles = Vec::with_capacity(NUM_CLIENTS);
32
33    for _ in 0..NUM_CLIENTS {
34        let addr = addr.to_string();
35        let client_sent = client_sent.clone();
36        let client_received = client_received.clone();
37        let client_connect_failed = client_connect_failed.clone();
38        let connect_gate = connect_gate.clone();
39
40        let handle = tokio::spawn(async move {
41            let permit = connect_gate.acquire_owned().await.unwrap();
42
43            let client = match Client::connect(&format!("ws://{}", addr)).await {
44                Ok(c) => c,
45                Err(_) => {
46                    client_connect_failed.fetch_add(1, Ordering::Relaxed);
47                    return None;
48                }
49            };
50
51            drop(permit);
52
53            {
54                let client_received = client_received.clone();
55                client.on("pong", move |_data| {
56                    let client_received = client_received.clone();
57                    async move {
58                        client_received.fetch_add(1, Ordering::Relaxed);
59                    }
60                });
61            }
62
63            for i in 0..MESSAGES_PER_CLIENT {
64                if client.emit("ping", format!("msg-{}", i)).await.is_ok() {
65                    client_sent.fetch_add(1, Ordering::Relaxed);
66                }
67            }
68
69            Some(client)
70        });
71
72        handles.push(handle);
73    }
74
75    let mut clients_kept_alive = Vec::with_capacity(NUM_CLIENTS);
76    for handle in handles {
77        if let Ok(Some(client)) = handle.await {
78            clients_kept_alive.push(client);
79        }
80    }
81
82    let sending_done_at = start.elapsed();
83
84    let expected_total = (NUM_CLIENTS * MESSAGES_PER_CLIENT) as u64;
85
86    let wait_timeout = Duration::from_secs(30);
87    let wait_start = Instant::now();
88
89    loop {
90        let received_now = client_received.load(Ordering::Relaxed);
91        if received_now >= expected_total || wait_start.elapsed() > wait_timeout {
92            break;
93        }
94        sleep(Duration::from_millis(50)).await;
95    }
96
97    let total_elapsed = start.elapsed();
98
99    println!("========== CLIENT-SIDE STATS ==========");
100    println!("Clients:                    {}", NUM_CLIENTS);
101    println!(
102        "Connect failures:           {}",
103        client_connect_failed.load(Ordering::Relaxed)
104    );
105    println!("Messages per client:        {}", MESSAGES_PER_CLIENT);
106    println!("Expected total messages:    {}", expected_total);
107    println!("------------------------------------------");
108    println!(
109        "Sent messages:               {}",
110        client_sent.load(Ordering::Relaxed)
111    );
112    println!(
113        "Replies received:            {}",
114        client_received.load(Ordering::Relaxed)
115    );
116    println!("------------------------------------------");
117    println!("Time to finish sending:     {:?}", sending_done_at);
118    println!("Total time (incl. replies): {:?}", total_elapsed);
119    println!(
120        "Client-side throughput:     {:.0} msgs/sec",
121        expected_total as f64 / total_elapsed.as_secs_f64()
122    );
123    println!("==========================================");
124}

examples/bench.rs (lines 77-82)

13async fn main() {
14    let addr = "127.0.0.1:9500";
15    let server = Server::bind(addr);
16
17    let server_received = Arc::new(AtomicU64::new(0));
18    let server_replied = Arc::new(AtomicU64::new(0));
19
20    {
21        let server_received = server_received.clone();
22        let server_replied = server_replied.clone();
23
24        server.on("ping", move |data, ctx| {
25            let server_received = server_received.clone();
26            let server_replied = server_replied.clone();
27            async move {
28                server_received.fetch_add(1, Ordering::Relaxed);
29                let text = data.parse::<String>().unwrap_or_default();
30                if ctx.emit("pong", text).await.is_ok() {
31                    server_replied.fetch_add(1, Ordering::Relaxed);
32                }
33            }
34        });
35    }
36
37    tokio::spawn(async move {
38        let _ = server.run().await;
39    });
40
41    sleep(Duration::from_millis(300)).await;
42
43    let client_sent = Arc::new(AtomicU64::new(0));
44    let client_received = Arc::new(AtomicU64::new(0));
45
46    let connect_gate = Arc::new(Semaphore::new(MAX_CONCURRENT_CONNECTS));
47
48    println!(
49        "Starting benchmark: {} clients x {} messages = {} total messages expected",
50        NUM_CLIENTS,
51        MESSAGES_PER_CLIENT,
52        NUM_CLIENTS * MESSAGES_PER_CLIENT
53    );
54
55    let start = Instant::now();
56
57    let mut handles = Vec::with_capacity(NUM_CLIENTS);
58
59    for _ in 0..NUM_CLIENTS {
60        let addr = addr.to_string();
61        let client_sent = client_sent.clone();
62        let client_received = client_received.clone();
63        let connect_gate = connect_gate.clone();
64
65        let handle = tokio::spawn(async move {
66            let permit = connect_gate.acquire_owned().await.unwrap();
67
68            let client = match Client::connect(&format!("ws://{}", addr)).await {
69                Ok(c) => c,
70                Err(_) => return None,
71            };
72
73            drop(permit);
74
75            {
76                let client_received = client_received.clone();
77                client.on("pong", move |_data| {
78                    let client_received = client_received.clone();
79                    async move {
80                        client_received.fetch_add(1, Ordering::Relaxed);
81                    }
82                });
83            }
84
85            for i in 0..MESSAGES_PER_CLIENT {
86                if client.emit("ping", format!("msg-{}", i)).await.is_ok() {
87                    client_sent.fetch_add(1, Ordering::Relaxed);
88                }
89            }
90
91            Some(client)
92        });
93
94        handles.push(handle);
95    }
96
97    let mut clients_kept_alive = Vec::with_capacity(NUM_CLIENTS);
98    for handle in handles {
99        if let Ok(Some(client)) = handle.await {
100            clients_kept_alive.push(client);
101        }
102    }
103
104    let sending_done_at = start.elapsed();
105
106    let expected_total = (NUM_CLIENTS * MESSAGES_PER_CLIENT) as u64;
107
108    let wait_timeout = Duration::from_secs(30);
109    let wait_start = Instant::now();
110
111    loop {
112        let received_now = client_received.load(Ordering::Relaxed);
113        if received_now >= expected_total || wait_start.elapsed() > wait_timeout {
114            break;
115        }
116        sleep(Duration::from_millis(50)).await;
117    }
118
119    let total_elapsed = start.elapsed();
120
121    println!();
122    println!("========== BENCHMARK RESULTS ==========");
123    println!("Clients spawned:            {}", NUM_CLIENTS);
124    println!("Messages per client:        {}", MESSAGES_PER_CLIENT);
125    println!("Expected total messages:    {}", expected_total);
126    println!("----------------------------------------");
127    println!(
128        "Client -> sent messages:    {}",
129        client_sent.load(Ordering::Relaxed)
130    );
131    println!(
132        "Server -> received:         {}",
133        server_received.load(Ordering::Relaxed)
134    );
135    println!(
136        "Server -> replies sent:     {}",
137        server_replied.load(Ordering::Relaxed)
138    );
139    println!(
140        "Client -> replies received: {}",
141        client_received.load(Ordering::Relaxed)
142    );
143    println!("----------------------------------------");
144    println!("Time to finish sending:     {:?}", sending_done_at);
145    println!("Total time (incl. replies): {:?}", total_elapsed);
146    println!(
147        "Throughput:                 {:.0} msgs/sec",
148        expected_total as f64 / total_elapsed.as_secs_f64()
149    );
150    println!("=========================================");
151}

Source

pub async fn emit<T: Serialize>(&self, event: &str, data: T) -> Result<()>

Sends an event to the server.

data is serialized to JSON and sent over the connection.

§Errors

Returns an error if data fails to serialize, or if the message could not be sent (e.g. the connection is closed).

§Example

use rustwave::{Client, Result};

    let client = Client::connect("ws://127.0.0.1:8080").await?;

    client.emit("message", "hello").await?;

    Ok(())

Examples found in repository ?

examples/test1.rs (line 11)

4async fn main() -> Result<()> {
5    let client = Client::connect("ws://127.0.0.1:8080/").await?;
6
7    client.on("pong", |data| async move {
8        println!("Server replied: {}", data);
9    });
10
11    client.emit("ping", "").await?;
12
13    tokio::time::sleep(std::time::Duration::from_secs(1)).await;
14
15    Ok(())
16}

More examples

Hide additional examples

examples/test2.rs (line 11)

4async fn main() -> Result<()> {
5    let client = Client::connect("ws://127.0.0.1:8080/chat").await?;
6
7    client.on("message", |data| async move {
8        println!("Got back: {}", data);
9    });
10
11    client.emit("mess", "hello from chat client").await?;
12
13    tokio::time::sleep(std::time::Duration::from_secs(1)).await;
14
15    client.wait().await;
16
17    Ok(())
18}

examples/terminal_client.rs (line 21)

13async fn main() -> Result<()> {
14    let client = Client::connect("ws://127.0.0.1:8080").await?;
15
16    client.on("chat", |data| async move {
17        println!("Room message:> {}", data);
18    });
19
20    let room = "chat";
21    client.emit("join-room", room).await?;
22
23    let mut stdin = BufReader::new(tokio::io::stdin());
24
25    loop {
26        print!("you> ");
27        std::io::stdout().flush().unwrap();
28
29        let mut line = String::new();
30        if stdin.read_line(&mut line).await.unwrap() == 0 {
31            break; 
32        }
33
34        let text = line.trim().to_string();
35        if text.is_empty() {
36            continue;
37        }
38
39        let msg = ChatMessage {
40            room: room.to_string(),
41            text,
42        };
43        client.emit("chat", msg).await.unwrap();
44    }
45
46    Ok(())
47}

examples/test3.rs (line 37)

14async fn main() -> Result<()> {
15    let client = Client::connect("ws://127.0.0.1:8080/game").await?;
16
17    client.on("online-count", |data| async move {
18        println!("Online in /game: {}", data);
19    });
20
21    client.on("player-joined", |data| async move {
22        println!("Someone joined the room: {}", data);
23    });
24
25    client.on("player-left", |data| async move {
26        println!("Someone left the room: {}", data);
27    });
28
29    client.on("typing", |data| async move {
30        println!("{}", data);
31    });
32
33    client.on("mess", |data| async move {
34        println!("Room message: {}", data);
35    });
36
37    client.emit("online-count", "").await?;
38    sleep(Duration::from_millis(300)).await;
39
40    client.emit("join-room", ROOM).await?;
41    sleep(Duration::from_millis(300)).await;
42
43    client.emit("typing", ROOM).await?;
44    sleep(Duration::from_millis(300)).await;
45
46    let msg = GameMessage {
47        mess: "hey room, I'm here!".to_string(),
48        room: ROOM.to_string(),
49    };
50    client.emit("mess", msg).await?;
51    sleep(Duration::from_millis(300)).await;
52
53    client.emit("leave-room", ROOM).await?;
54
55    sleep(Duration::from_secs(3)).await;
56
57    Ok(())
58}

examples/client_banch.rs (line 64)

13async fn main() {
14    let addr = "127.0.0.1:9500";
15
16    let client_sent = Arc::new(AtomicU64::new(0));
17    let client_received = Arc::new(AtomicU64::new(0));
18    let client_connect_failed = Arc::new(AtomicU64::new(0));
19
20    let connect_gate = Arc::new(Semaphore::new(MAX_CONCURRENT_CONNECTS));
21
22    println!(
23        "Starting client benchmark: {} clients x {} messages = {} total messages expected",
24        NUM_CLIENTS,
25        MESSAGES_PER_CLIENT,
26        NUM_CLIENTS * MESSAGES_PER_CLIENT
27    );
28
29    let start = Instant::now();
30
31    let mut handles = Vec::with_capacity(NUM_CLIENTS);
32
33    for _ in 0..NUM_CLIENTS {
34        let addr = addr.to_string();
35        let client_sent = client_sent.clone();
36        let client_received = client_received.clone();
37        let client_connect_failed = client_connect_failed.clone();
38        let connect_gate = connect_gate.clone();
39
40        let handle = tokio::spawn(async move {
41            let permit = connect_gate.acquire_owned().await.unwrap();
42
43            let client = match Client::connect(&format!("ws://{}", addr)).await {
44                Ok(c) => c,
45                Err(_) => {
46                    client_connect_failed.fetch_add(1, Ordering::Relaxed);
47                    return None;
48                }
49            };
50
51            drop(permit);
52
53            {
54                let client_received = client_received.clone();
55                client.on("pong", move |_data| {
56                    let client_received = client_received.clone();
57                    async move {
58                        client_received.fetch_add(1, Ordering::Relaxed);
59                    }
60                });
61            }
62
63            for i in 0..MESSAGES_PER_CLIENT {
64                if client.emit("ping", format!("msg-{}", i)).await.is_ok() {
65                    client_sent.fetch_add(1, Ordering::Relaxed);
66                }
67            }
68
69            Some(client)
70        });
71
72        handles.push(handle);
73    }
74
75    let mut clients_kept_alive = Vec::with_capacity(NUM_CLIENTS);
76    for handle in handles {
77        if let Ok(Some(client)) = handle.await {
78            clients_kept_alive.push(client);
79        }
80    }
81
82    let sending_done_at = start.elapsed();
83
84    let expected_total = (NUM_CLIENTS * MESSAGES_PER_CLIENT) as u64;
85
86    let wait_timeout = Duration::from_secs(30);
87    let wait_start = Instant::now();
88
89    loop {
90        let received_now = client_received.load(Ordering::Relaxed);
91        if received_now >= expected_total || wait_start.elapsed() > wait_timeout {
92            break;
93        }
94        sleep(Duration::from_millis(50)).await;
95    }
96
97    let total_elapsed = start.elapsed();
98
99    println!("========== CLIENT-SIDE STATS ==========");
100    println!("Clients:                    {}", NUM_CLIENTS);
101    println!(
102        "Connect failures:           {}",
103        client_connect_failed.load(Ordering::Relaxed)
104    );
105    println!("Messages per client:        {}", MESSAGES_PER_CLIENT);
106    println!("Expected total messages:    {}", expected_total);
107    println!("------------------------------------------");
108    println!(
109        "Sent messages:               {}",
110        client_sent.load(Ordering::Relaxed)
111    );
112    println!(
113        "Replies received:            {}",
114        client_received.load(Ordering::Relaxed)
115    );
116    println!("------------------------------------------");
117    println!("Time to finish sending:     {:?}", sending_done_at);
118    println!("Total time (incl. replies): {:?}", total_elapsed);
119    println!(
120        "Client-side throughput:     {:.0} msgs/sec",
121        expected_total as f64 / total_elapsed.as_secs_f64()
122    );
123    println!("==========================================");
124}

examples/bench.rs (line 86)

13async fn main() {
14    let addr = "127.0.0.1:9500";
15    let server = Server::bind(addr);
16
17    let server_received = Arc::new(AtomicU64::new(0));
18    let server_replied = Arc::new(AtomicU64::new(0));
19
20    {
21        let server_received = server_received.clone();
22        let server_replied = server_replied.clone();
23
24        server.on("ping", move |data, ctx| {
25            let server_received = server_received.clone();
26            let server_replied = server_replied.clone();
27            async move {
28                server_received.fetch_add(1, Ordering::Relaxed);
29                let text = data.parse::<String>().unwrap_or_default();
30                if ctx.emit("pong", text).await.is_ok() {
31                    server_replied.fetch_add(1, Ordering::Relaxed);
32                }
33            }
34        });
35    }
36
37    tokio::spawn(async move {
38        let _ = server.run().await;
39    });
40
41    sleep(Duration::from_millis(300)).await;
42
43    let client_sent = Arc::new(AtomicU64::new(0));
44    let client_received = Arc::new(AtomicU64::new(0));
45
46    let connect_gate = Arc::new(Semaphore::new(MAX_CONCURRENT_CONNECTS));
47
48    println!(
49        "Starting benchmark: {} clients x {} messages = {} total messages expected",
50        NUM_CLIENTS,
51        MESSAGES_PER_CLIENT,
52        NUM_CLIENTS * MESSAGES_PER_CLIENT
53    );
54
55    let start = Instant::now();
56
57    let mut handles = Vec::with_capacity(NUM_CLIENTS);
58
59    for _ in 0..NUM_CLIENTS {
60        let addr = addr.to_string();
61        let client_sent = client_sent.clone();
62        let client_received = client_received.clone();
63        let connect_gate = connect_gate.clone();
64
65        let handle = tokio::spawn(async move {
66            let permit = connect_gate.acquire_owned().await.unwrap();
67
68            let client = match Client::connect(&format!("ws://{}", addr)).await {
69                Ok(c) => c,
70                Err(_) => return None,
71            };
72
73            drop(permit);
74
75            {
76                let client_received = client_received.clone();
77                client.on("pong", move |_data| {
78                    let client_received = client_received.clone();
79                    async move {
80                        client_received.fetch_add(1, Ordering::Relaxed);
81                    }
82                });
83            }
84
85            for i in 0..MESSAGES_PER_CLIENT {
86                if client.emit("ping", format!("msg-{}", i)).await.is_ok() {
87                    client_sent.fetch_add(1, Ordering::Relaxed);
88                }
89            }
90
91            Some(client)
92        });
93
94        handles.push(handle);
95    }
96
97    let mut clients_kept_alive = Vec::with_capacity(NUM_CLIENTS);
98    for handle in handles {
99        if let Ok(Some(client)) = handle.await {
100            clients_kept_alive.push(client);
101        }
102    }
103
104    let sending_done_at = start.elapsed();
105
106    let expected_total = (NUM_CLIENTS * MESSAGES_PER_CLIENT) as u64;
107
108    let wait_timeout = Duration::from_secs(30);
109    let wait_start = Instant::now();
110
111    loop {
112        let received_now = client_received.load(Ordering::Relaxed);
113        if received_now >= expected_total || wait_start.elapsed() > wait_timeout {
114            break;
115        }
116        sleep(Duration::from_millis(50)).await;
117    }
118
119    let total_elapsed = start.elapsed();
120
121    println!();
122    println!("========== BENCHMARK RESULTS ==========");
123    println!("Clients spawned:            {}", NUM_CLIENTS);
124    println!("Messages per client:        {}", MESSAGES_PER_CLIENT);
125    println!("Expected total messages:    {}", expected_total);
126    println!("----------------------------------------");
127    println!(
128        "Client -> sent messages:    {}",
129        client_sent.load(Ordering::Relaxed)
130    );
131    println!(
132        "Server -> received:         {}",
133        server_received.load(Ordering::Relaxed)
134    );
135    println!(
136        "Server -> replies sent:     {}",
137        server_replied.load(Ordering::Relaxed)
138    );
139    println!(
140        "Client -> replies received: {}",
141        client_received.load(Ordering::Relaxed)
142    );
143    println!("----------------------------------------");
144    println!("Time to finish sending:     {:?}", sending_done_at);
145    println!("Total time (incl. replies): {:?}", total_elapsed);
146    println!(
147        "Throughput:                 {:.0} msgs/sec",
148        expected_total as f64 / total_elapsed.as_secs_f64()
149    );
150    println!("=========================================");
151}

Source

pub async fn close(&self)

Closes the WebSocket connection.

§Example

use rustwave::{Client, Result};

    let client = Client::connect("ws://127.0.0.1:8080").await?;

    client.close().await;
    Ok(())

Source

pub async fn wait(&self)

Waits until the client’s connection has been closed.

This resolves once the underlying connection has shut down — whether due to calling Client::close, the server closing the connection, a heartbeat timeout, or a connection error. Useful for keeping a program alive while the client runs in the background.

§Example

use rustwave::{Client, Result};

#[tokio::main]
async fn main() -> Result<()> {
    let client = Client::connect("ws://127.0.0.1:8080").await?;

    client.on("message", |data| async move {
        println!("Server say: {}", data);
    });

   
    client.wait().await;
    Ok(())
}

Examples found in repository ?

examples/test2.rs (line 15)

4async fn main() -> Result<()> {
5    let client = Client::connect("ws://127.0.0.1:8080/chat").await?;
6
7    client.on("message", |data| async move {
8        println!("Got back: {}", data);
9    });
10
11    client.emit("mess", "hello from chat client").await?;
12
13    tokio::time::sleep(std::time::Duration::from_secs(1)).await;
14
15    client.wait().await;
16
17    Ok(())
18}