use rustwave::{Client, RustWaveError, Server};
use std::time::Duration;
use tokio::{sync::mpsc::channel, time::sleep};
use uuid::Uuid;
#[tokio::test]
async fn test_server_echoes_message() {
let addr = "127.0.0.1:9001";
let server = Server::bind(addr);
server.on("ping", |data, ctx| async move {
let text = data.parse::<String>().unwrap_or_default();
let _ = ctx.emit("pong", format!("echo: {}", text)).await;
});
tokio::spawn(async move {
let _ = server.run().await;
});
sleep(Duration::from_millis(200)).await;
let client = Client::connect(&format!("ws://{}", addr)).await.unwrap();
let (tx, mut rx) = channel::<String>(1);
client.on("pong", move |data| {
let tx = tx.clone();
async move {
let text = data.parse::<String>().unwrap_or_default();
let _ = tx.send(text).await;
}
});
client.emit("ping", "hello").await.unwrap();
let result = rx.recv().await.unwrap();
assert_eq!(result, "echo: hello");
}
#[tokio::test]
async fn test_room_broadcast_only_reaches_room_members() {
let addr = "127.0.0.1:9002";
let server = Server::bind(addr);
server.on("connection", |_data, ctx| async move {
ctx.join_room("room-1");
});
server.on("message", |data, ctx| async move {
let text = data.parse::<String>().unwrap_or_default();
let _ = ctx.broadcast_to_room("room-1", "room-message", text).await;
});
tokio::spawn(async move {
let _ = server.run().await;
});
sleep(Duration::from_millis(200)).await;
let client_a = Client::connect(&format!("ws://{}", addr)).await.unwrap();
let client_b = Client::connect(&format!("ws://{}", addr)).await.unwrap();
let (tx_a, mut rx_a) = channel::<String>(5);
client_a.on("room-message", move |data| {
let tx = tx_a.clone();
async move {
let _ = tx.send(data.parse::<String>().unwrap_or_default()).await;
}
});
sleep(Duration::from_millis(200)).await;
client_b.emit("message", "hi room").await.unwrap();
let result = rx_a.recv().await.unwrap();
assert_eq!(result, "hi room");
}
#[tokio::test]
async fn test_emit_to_unknown_socket_returns_client_not_found() {
let addr = "127.0.0.1:9201";
let server = Server::bind(addr);
let root = server.namespace("/");
tokio::spawn(async move {
let _ = server.run().await;
});
sleep(Duration::from_millis(200)).await;
let fake_id = Uuid::new_v4();
let result = root.emit_to(&fake_id, "hello", "test").await;
assert!(result.is_err());
match result.unwrap_err() {
RustWaveError::ClientNotFound(id) => {
assert_eq!(id, fake_id.to_string());
}
other => panic!("expected ClientNotFound, got {:?}", other),
}
}
#[tokio::test]
async fn test_broadcast_to_expect_excludes_sender() {
let addr = "127.0.0.1:9202";
let server = Server::bind(addr);
server.on("connection", |_data, ctx| async move {
ctx.join_room("room-1");
});
server.on("message", |data, ctx| async move {
let text = data.parse::<String>().unwrap_or_default();
let sender_id = ctx.get_socket_id();
let _ = ctx
.broadcast_to_expect("room-1", sender_id, "room-message", text)
.await;
});
tokio::spawn(async move {
let _ = server.run().await;
});
sleep(Duration::from_millis(200)).await;
let client_a = Client::connect(&format!("ws://{}", addr)).await.unwrap();
let client_b = Client::connect(&format!("ws://{}", addr)).await.unwrap();
let (tx_a, mut rx_a) = channel::<String>(1);
client_a.on("room-message", move |data| {
let tx = tx_a.clone();
async move {
let _ = tx.send(data.parse::<String>().unwrap_or_default()).await;
}
});
let (tx_b, mut rx_b) = channel::<String>(1);
client_b.on("room-message", move |data| {
let tx = tx_b.clone();
async move {
let _ = tx.send(data.parse::<String>().unwrap_or_default()).await;
}
});
sleep(Duration::from_millis(200)).await;
client_a.emit("message", "hi everyone").await.unwrap();
let result_b = tokio::time::timeout(Duration::from_secs(2), rx_b.recv())
.await
.expect("client_b never receive the broadcast")
.unwrap();
assert_eq!(result_b, "hi everyone");
let result_a = tokio::time::timeout(Duration::from_millis(500), rx_a.recv()).await;
assert!(
result_a.is_err(),
"sender should have been excluded but received the broadcast anyway"
);
}
#[tokio::test]
async fn test_namespace_isolation() {
let addr = "127.0.0.1:9203";
let server = Server::bind(addr);
let chat = server.namespace("/chat");
let (root_tx, mut root_rx) = channel::<String>(5);
server.on("message", move |data, _ctx| {
let tx = root_tx.clone();
async move {
let _ = tx.send(data.parse::<String>().unwrap_or_default()).await;
}
});
let (chat_tx, mut chat_rx) = channel::<String>(5);
chat.on("message", move |data, _ctx| {
let tx = chat_tx.clone();
async move {
let _ = tx.send(data.parse::<String>().unwrap_or_default()).await;
}
});
tokio::spawn(async move {
let _ = server.run().await;
});
sleep(Duration::from_millis(200)).await;
let root_client = Client::connect(&format!("ws://{}/", addr)).await.unwrap();
let chat_client = Client::connect(&format!("ws://{}/chat", addr))
.await
.unwrap();
sleep(Duration::from_millis(200)).await;
root_client.emit("message", "from root").await.unwrap();
chat_client.emit("message", "from chat").await.unwrap();
let root_result = tokio::time::timeout(Duration::from_secs(2), root_rx.recv())
.await
.expect("root handler never fired")
.unwrap();
assert_eq!(root_result, "from root");
let chat_result = tokio::time::timeout(Duration::from_secs(2), chat_rx.recv())
.await
.expect("chat handler never fired")
.unwrap();
assert_eq!(chat_result, "from chat");
let extra_on_root = tokio::time::timeout(Duration::from_millis(300), root_rx.recv()).await;
assert!(
extra_on_root.is_err(),
"root received an unexpected extra message (possible leak from /chat)"
);
let extra_on_chat = tokio::time::timeout(Duration::from_millis(300), chat_rx.recv()).await;
assert!(
extra_on_chat.is_err(),
"/chat received an unexpected extra message (possible leak from root)"
);
}