use axum::{ Router, extract::{Query, State}, routing::get, }; use mirror_tool::{ config::{ScoreboardConfig, ScoreboardKind}, scoreboard::Scoreboard, }; use serde_json::json; use std::{ collections::HashMap, sync::{ Arc, atomic::{AtomicUsize, Ordering::SeqCst}, }, time::Duration, }; use tokio::sync::Mutex; use tokio_util::sync::CancellationToken; #[derive(Default)] struct Mock { handshakes: AtomicUsize, polls: AtomicUsize, posts: Mutex>, } async fn poll( State(mock): State>, Query(query): Query>, ) -> String { assert_eq!(query.get("EIO").map(String::as_str), Some("4")); assert_eq!(query.get("transport").map(String::as_str), Some("polling")); if !query.contains_key("sid") { let number = mock.handshakes.fetch_add(1, SeqCst) + 1; return format!( "0{}", json!({"sid":format!("test_{number}"),"pingInterval":1000,"pingTimeout":1000,"upgrades":[],"maxPayload":1000000}) ); } let call = mock.polls.fetch_add(1, SeqCst); if call == 0 { format!( "40/game_events,{{\"sid\":\"namespace-test\"}}\u{1e}42/game_events,{}", json!(["init_scoreboard",{"data":{"teams":[{"id":1,"name":"A"}],"tasks":[{"id":2,"name":"Notes"}],"state":{"round":1,"team_tasks":[{"team_id":1,"task_id":2,"status":101}]}}}]) ) } else if call == 1 { "2heartbeat".into() } else if call == 2 { format!( "42/game_events,{}", json!(["update_scoreboard",{"data":{"round":2,"team_tasks":[{"team_id":1,"task_id":2,"status":104,"stolen":7}]}}]) ) } else if call == 3 { tokio::time::sleep(Duration::from_millis(200)).await; "1".into() } else if query.get("sid").is_some_and(|s| s == "test_2") { tokio::time::sleep(Duration::from_millis(50)).await; format!( "40/game_events,{{}}\u{1e}42/game_events,{}", json!(["init_scoreboard",{"data":{"teams":[{"id":1,"name":"A"}],"tasks":[{"id":2,"name":"Notes"}],"state":{"round":3,"team_tasks":[{"team_id":1,"task_id":2,"status":101}]}}}]) ) } else { tokio::time::sleep(Duration::from_millis(50)).await; "2".into() } } async fn post(State(mock): State>, body: String) -> &'static str { mock.posts.lock().await.push(body); "ok" } #[tokio::test] async fn actual_polling_handshake_ping_delta_and_reconnect() { let mock = Arc::new(Mock::default()); let router = Router::new() .route("/socket.io/", get(poll).post(post)) .with_state(mock.clone()); let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let addr = listener.local_addr().unwrap(); let server = tokio::spawn(async move { axum::serve(listener, router).await.unwrap(); }); let cfg = ScoreboardConfig { kind: ScoreboardKind::Forcad, url: format!("http://{addr}"), poll_interval_ms: 100, stale_after_ms: 2000, }; let board = Scoreboard::new(Some(&cfg)); let cancel = CancellationToken::new(); let worker = tokio::spawn(board.clone().run(cfg, cancel.clone())); tokio::time::timeout(Duration::from_secs(3), async { loop { if board.snapshot().round == 2 { break; } tokio::time::sleep(Duration::from_millis(5)).await; } }) .await .unwrap(); let delta = board.snapshot(); assert_eq!(delta.teams[0].name, "A"); assert_eq!(delta.entries["2"]["1"].stolen, 7); assert!(mock.posts.lock().await.iter().any(|s| s == "3heartbeat")); tokio::time::timeout(Duration::from_secs(4), async { loop { if board.snapshot().round == 3 { break; } tokio::time::sleep(Duration::from_millis(10)).await; } }) .await .unwrap(); assert_eq!(mock.handshakes.load(SeqCst), 2); cancel.cancel(); worker.await.unwrap(); assert!(!board.snapshot().connected); server.abort(); let _ = server.await; } #[tokio::test] async fn continuous_non_ping_packets_cannot_extend_negotiated_heartbeat() { let handshakes = Arc::new(AtomicUsize::new(0)); let polls = Arc::new(AtomicUsize::new(0)); let state = (handshakes.clone(), polls.clone()); let router = Router::new() .route( "/socket.io/", get( |State((handshakes, polls)): State<(Arc, Arc)>, Query(query): Query>| async move { if !query.contains_key("sid") { handshakes.fetch_add(1, SeqCst); return format!( "0{}", json!({"sid":"no_ping", "pingInterval":500, "pingTimeout":500}) ); } polls.fetch_add(1, SeqCst); tokio::time::sleep(Duration::from_millis(10)).await; "40/game_events,{}".into() }, ) .post(|| async { "ok" }), ) .with_state(state); let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let addr = listener.local_addr().unwrap(); let server = tokio::spawn(async move { axum::serve(listener, router).await.unwrap(); }); let cfg = ScoreboardConfig { kind: ScoreboardKind::Forcad, url: format!("http://{addr}"), poll_interval_ms: 100, stale_after_ms: 2000, }; let board = Scoreboard::new(Some(&cfg)); let cancel = CancellationToken::new(); let worker = tokio::spawn(board.run(cfg, cancel.clone())); let reconnected = tokio::time::timeout(Duration::from_secs(4), async { while handshakes.load(SeqCst) < 2 { tokio::time::sleep(Duration::from_millis(10)).await; } }) .await; cancel.cancel(); worker.await.unwrap(); server.abort(); let _ = server.await; assert!( reconnected.is_ok(), "non-ping frames kept old session alive" ); assert!( polls.load(SeqCst) > 10, "fixture did not continuously return frames" ); }