diff options
Diffstat (limited to 'server/replaytool/src/main.rs')
| -rw-r--r-- | server/replaytool/src/main.rs | 173 |
1 files changed, 13 insertions, 160 deletions
diff --git a/server/replaytool/src/main.rs b/server/replaytool/src/main.rs index 337152ec..5ff821cc 100644 --- a/server/replaytool/src/main.rs +++ b/server/replaytool/src/main.rs @@ -15,24 +15,21 @@ along with this program. If not, see <https://www.gnu.org/licenses/>. */ -use anyhow::{anyhow, Context}; -use async_compression::tokio::{bufread::ZstdDecoder, write::ZstdEncoder}; + +pub mod record; +pub mod replay; + use clap::Parser; -use futures_util::{SinkExt, StreamExt}; -use hurrycurry_protocol::{Message, PacketC, PacketS}; -use log::{debug, info, warn, LevelFilter}; +use hurrycurry_protocol::PacketC; +use log::{info, warn, LevelFilter}; use serde::{Deserialize, Serialize}; use std::{ - path::{Path, PathBuf}, - time::{Duration, Instant, SystemTime}, -}; -use tokio::{ - fs::File, - io::{AsyncBufReadExt, AsyncWriteExt, BufReader, BufWriter}, - net::TcpListener, - time::sleep, + path::PathBuf, + time::{Duration, SystemTime}, }; -use tokio_tungstenite::tungstenite; +use tokio::{net::TcpListener, time::sleep}; + +use crate::{record::record, replay::replay}; #[derive(Parser)] enum Args { @@ -84,7 +81,7 @@ async fn main() -> anyhow::Result<()> { } else { output.clone() }; - if let Err(e) = do_record(&out, &url).await { + if let Err(e) = record(&out, &url).await { warn!("recording failed: {e}"); sleep(Duration::from_secs(1)).await; } @@ -99,153 +96,9 @@ async fn main() -> anyhow::Result<()> { info!("listening for websockets on {}", ws_listener.local_addr()?); loop { - let mut file = - BufReader::new(ZstdDecoder::new(BufReader::new(File::open(&input).await?))) - .lines(); - - let mut next_event = - serde_json::from_str::<Event>(&file.next_line().await?.ok_or(anyhow!("eof"))?)?; - let mut time = 0.; - let mut paused = false; - - info!("ready"); - let (sock, addr) = ws_listener.accept().await?; - let Ok(mut sock) = tokio_tungstenite::accept_async(sock).await else { - warn!("invalid ws handshake"); - continue; - }; - info!("{addr} connected via ws"); - - sock.send(tokio_tungstenite::tungstenite::Message::Text( - serde_json::to_string(&PacketC::ReplayStart).unwrap().into(), - )) - .await?; - while let Some(Ok(message)) = sock.next().await { - let mut send = async |p: PacketC| { - sock.send(tokio_tungstenite::tungstenite::Message::Text( - serde_json::to_string(&p).unwrap().into(), - )) - .await - }; - match message { - tungstenite::Message::Text(line) => { - let packet: PacketS = match serde_json::from_str(&line) { - Ok(p) => p, - Err(e) => { - warn!("invalid packet: {e}"); - break; - } - }; - debug!("<- {packet:?}"); - - match packet { - PacketS::Join { .. } => { - send(PacketC::ServerMessage { - message: Message::Translation { - id: "s.replay.cannot_join".to_owned(), - params: vec![], - }, - error: true, - }) - .await?; - } - PacketS::Idle { paused: pause } => { - // Cannot display server hint like in main server because cant address them to spectators. - paused = pause; - send(PacketC::Pause { state: pause }).await? - } - PacketS::ReplayTick { dt } => { - if paused { - continue; - } - time += dt; - while next_event.ts < time { - debug!("<- {:?}", next_event.packet); - if !matches!( - next_event.packet, - PacketC::SetIngame { state: false, .. } - ) { - send(next_event.packet).await?; - } - - if let Some(next) = &file.next_line().await? { - next_event = serde_json::from_str::<Event>(next)?; - } else { - info!("reached end"); - send(PacketC::ServerMessage { - message: Message::Text("End of Replay".to_string()), - error: false, - }) - .await?; - send(PacketC::ReplayStop).await?; - next_event = Event { - ts: f64::INFINITY, - packet: PacketC::FlushMap, - }; - continue; - }; - } - } - x => warn!("unhandled client packet: {x:?}"), - } - } - tungstenite::Message::Close(_) => break, - _ => (), - } - } - info!("{addr} left"); - } - } - } - Ok(()) -} - -pub async fn do_record(output: &Path, url: &str) -> anyhow::Result<()> { - let mut file = BufWriter::new(ZstdEncoder::new(BufWriter::new( - File::create(&output).await?, - ))); - info!("connecting to {url:?}..."); - let (mut sock, _) = tokio_tungstenite::connect_async(url).await?; - info!("starting recording."); - let start = Instant::now(); - - while let Some(Ok(message)) = sock.next().await { - match message { - tungstenite::Message::Text(line) => { - let packet: PacketC = serde_json::from_str(&line).context("invalid packet")?; - debug!("<- {packet:?}"); - - let is_end = matches!(packet, PacketC::SetIngame { state: false, .. }); - - file.write_all( - format!( - "{}\n", - serde_json::to_string(&Event { - ts: start.elapsed().as_secs_f64(), - packet - }) - .unwrap() - ) - .as_bytes(), - ) - .await?; - - if is_end { - info!("stopping replay..."); - break; - } + replay(&ws_listener, &input).await?; } - tungstenite::Message::Close(_) => break, - _ => (), } } - file.flush().await?; - let mut file = file.into_inner(); - file.flush().await?; - let mut file = file.into_inner(); - file.flush().await?; - let mut file = file.into_inner(); - file.flush().await?; - info!("done"); Ok(()) } |