From e1e2e576e29abb3560483fcd5e2ecf7f3f663a84 Mon Sep 17 00:00:00 2001 From: Sebastian Moser Date: Tue, 18 Apr 2023 18:21:56 +0200 Subject: [PATCH] implemented Ping Message response and some fixes --- src/server.rs | 11 +++++------ src/server/proto.rs | 6 ++++-- 2 files changed, 9 insertions(+), 8 deletions(-) diff --git a/src/server.rs b/src/server.rs index d12b21f..9fc8b7b 100644 --- a/src/server.rs +++ b/src/server.rs @@ -241,7 +241,7 @@ pub async fn run_server(args: Vec) { // the update handling task let update_mutexes = mutexes.clone(); tokio::spawn(async move { - proto::update_thread(update_rx, update_mutexes); + proto::update_thread(update_rx, update_mutexes).await.handle().extra_msg("update thread crashed").is_critical(); }); @@ -372,6 +372,9 @@ async fn handle_websocket_connection( proto::MizeMessage::Bin(bin_msg) => { socket_tx.send(Message::Binary(bin_msg.raw)).await; } + proto::MizeMessage::Pong() => { + socket_tx.send(Message::Pong(Vec::new())).await; + } } } }); @@ -475,11 +478,7 @@ async fn handle_websocket_connection( Message::Ping(_) => { - let err = MizeError::new(11) - .extra_msg("unhandeld WebSocket-Message type: Ping") - .handle(); - - msg_tx.send(proto::MizeMessage::Json(err.to_json_message())).await; + msg_tx.send(MizeMessage::Pong()); }, diff --git a/src/server/proto.rs b/src/server/proto.rs index 1379953..4650f3c 100644 --- a/src/server/proto.rs +++ b/src/server/proto.rs @@ -32,6 +32,7 @@ use serde::{Serialize, Deserialize}; pub enum MizeMessage { Json(JsonMessage), Bin(BinMessage), + Pong(), } @@ -160,7 +161,6 @@ pub struct UpdateMessage { #[derive(Clone, Debug, Serialize, Deserialize)] pub struct Delta { - #[serde(flatten)] pub delta: Vec<(Path, JsonValue)>, } @@ -386,12 +386,14 @@ pub async fn handle_json_msg(msg: JsonMessage, origin: Peer, mutexes: Mutexes) - } pub async fn handle_update(update: Update, origin: Peer, mutexes: Mutexes) -> Result<(), MizeError> { - mutexes.update_tx.send((update, origin)); + mutexes.update_tx.send((update, origin)).await.map_err(|err| MizeError::new(11).extra_msg(format!("From SendError: {}", err))); Ok(()) } pub async fn update_thread(mut update_rx: mpsc::Receiver<(Update, Peer)>, mutexes: Mutexes) -> Result<(), MizeError> { while let Some((update, origin)) = update_rx.recv().await { + println!("UPDATE: {:?}", update.iter().next().unwrap().0); + println!("DELTA: {:?}", update.iter().next().unwrap().1); //TODO: run update code from modules and types let itemstore = mutexes.itemstore.lock().await;