use axum::extract::ws::{Message, WebSocket, WebSocketUpgrade};
use std::sync::Arc;
pub trait WsHandler: Send + Sync + 'static {
fn on_connect(&self) {}
fn on_message(&self, _msg: Message) -> Option<Message> {
None
}
fn on_close(&self) {}
}
pub fn ws_handler<H: WsHandler>(handler: H) -> axum::routing::MethodRouter<()> {
let handler = Arc::new(handler);
axum::routing::get(move |ws: WebSocketUpgrade| async move {
let handler = handler.clone();
ws.on_upgrade(move |socket| handle_ws_connection(socket, handler))
})
}
async fn handle_ws_connection(mut socket: WebSocket, handler: Arc<dyn WsHandler>) {
handler.on_connect();
while let Some(Ok(msg)) = socket.recv().await {
if matches!(msg, Message::Close(_)) {
break;
}
if let Some(reply) = handler.on_message(msg) {
if socket.send(reply).await.is_err() {
break;
}
}
}
handler.on_close();
}
#[derive(Debug, Default, Clone)]
pub struct EchoWsHandler;
impl EchoWsHandler {
pub fn new() -> Self {
Self
}
}
impl WsHandler for EchoWsHandler {
fn on_message(&self, msg: Message) -> Option<Message> {
Some(msg)
}
}
#[cfg(test)]
mod tests {
use super::*;
use axum::body::Body;
use axum::http::{Method, Request, StatusCode};
use tower::ServiceExt;
#[test]
fn test_echo_handler_returns_message() {
let handler = EchoWsHandler::new();
let msg = Message::text("hello");
let result = handler.on_message(msg);
assert!(result.is_some());
}
#[test]
fn test_echo_handler_default() {
let handler = EchoWsHandler;
let msg = Message::text("test");
assert!(handler.on_message(msg).is_some());
}
struct NoReplyHandler;
impl WsHandler for NoReplyHandler {
fn on_message(&self, _msg: Message) -> Option<Message> {
None
}
}
#[test]
fn test_custom_handler_no_reply() {
let handler = NoReplyHandler;
let msg = Message::text("hello");
assert!(handler.on_message(msg).is_none());
}
struct PrefixHandler;
impl WsHandler for PrefixHandler {
fn on_message(&self, _msg: Message) -> Option<Message> {
Some(Message::text("prefix: reply"))
}
}
#[test]
fn test_custom_handler_with_reply() {
let handler = PrefixHandler;
let msg = Message::text("input");
let reply = handler.on_message(msg).unwrap();
assert_eq!(reply.to_text().unwrap(), "prefix: reply");
}
#[test]
fn test_default_lifecycle_hooks_no_panic() {
let handler = EchoWsHandler::new();
handler.on_connect();
handler.on_close();
}
#[tokio::test]
async fn test_ws_route_registered_as_get() {
let router = axum::Router::new().route("/ws/echo", ws_handler(EchoWsHandler::new()));
let request = Request::builder()
.method(Method::GET)
.uri("/ws/echo")
.body(Body::empty())
.unwrap();
let response = router.oneshot(request).await.unwrap();
assert_eq!(response.status(), StatusCode::BAD_REQUEST);
}
#[tokio::test]
async fn test_ws_route_not_found() {
let router = axum::Router::new().route("/ws/echo", ws_handler(EchoWsHandler::new()));
let request = Request::builder()
.method(Method::GET)
.uri("/ws/nonexistent")
.body(Body::empty())
.unwrap();
let response = router.oneshot(request).await.unwrap();
assert_eq!(response.status(), StatusCode::NOT_FOUND);
}
#[tokio::test]
async fn test_ws_route_rejects_post() {
let router = axum::Router::new().route("/ws/echo", ws_handler(EchoWsHandler::new()));
let request = Request::builder()
.method(Method::POST)
.uri("/ws/echo")
.body(Body::empty())
.unwrap();
let response = router.oneshot(request).await.unwrap();
assert_eq!(response.status(), StatusCode::METHOD_NOT_ALLOWED);
}
}