From 26316e6f47ed604ea6c36f514c1dbe84281da18e Mon Sep 17 00:00:00 2001 From: Thilo Behnke Date: Sat, 11 Jun 2022 14:09:24 +0200 Subject: [PATCH] close session on websocket closed event --- server/src/http.rs | 51 ++++++++++++++++++++++++++++++++++++--------- server/src/utils.rs | 12 +++++++++++ 2 files changed, 53 insertions(+), 10 deletions(-) diff --git a/server/src/http.rs b/server/src/http.rs index 675a31b..7815825 100644 --- a/server/src/http.rs +++ b/server/src/http.rs @@ -4,11 +4,11 @@ use std::io::ErrorKind::NotFound; use std::net::SocketAddr; use std::str::FromStr; use std::sync::Arc; -use std::time::Duration; +use std::time::{Duration, SystemTime}; use hyper::{Body, body, Method, Request, Response, Server, StatusCode}; use hyper::server::conn::AddrStream; use hyper::service::{make_service_fn, service_fn}; -use hyper_tungstenite::HyperWebsocket; +use hyper_tungstenite::{HyperWebsocket, tungstenite}; use hyper_tungstenite::tungstenite::{Error, Message}; use kafka::producer::Producer; use serde_json::json; @@ -16,11 +16,14 @@ use serde::{Deserialize, Serialize}; use tokio::sync::Mutex; use pong::event::event::{Event, EventReader, EventWriter}; use futures::{sink::SinkExt, stream::StreamExt}; +use tokio::io::Sink; +use tokio::task; use tokio::time::sleep; use crate::kafka::{KafkaEventReaderImpl, KafkaSessionEventWriterImpl}; use crate::player::Player; use crate::session::{Session, SessionManager}; use crate::utils::http_utils::{get_query_params, read_json_body}; +use crate::utils::time_utils::now; pub struct HttpServer { addr: [u8; 4], @@ -62,12 +65,14 @@ impl HttpServer { eprintln!("{}", error); return build_error_res(error.as_str(), StatusCode::BAD_REQUEST); } - if let None = session_manager.lock().await.get_session(session_id) { + let session = session_manager.lock().await.get_session(session_id); + if let None = session { let error = format!("Session does not exist: {}", session_id); eprintln!("{}", error); return build_error_res(error.as_str(), StatusCode::NOT_FOUND); } - let websocket_session = WebSocketSession {session_id: session_id.to_string(), connection_type: connection_type.unwrap()}; + let session = session.unwrap(); + let websocket_session = WebSocketSession {session: session.clone(), connection_type: connection_type.unwrap()}; println!("Websocket upgrade request is valid, will now upgrade to websocket: {:?}", req); let (response, websocket) = hyper_tungstenite::upgrade(req, None).unwrap(); @@ -104,7 +109,8 @@ async fn serve_websocket(websocket_session: WebSocketSession, websocket: HyperWe let (mut websocket_writer, mut websocket_reader) = websocket.split(); let session_manager = session_manager.lock().await; - let event_handler_pair = session_manager.split(&websocket_session.session_id, websocket_session.connection_type.get_topics()); + let session = session_manager.get_session(&websocket_session.session.hash); + let event_handler_pair = session_manager.split(&websocket_session.session.hash, websocket_session.connection_type.get_topics()); if let Err(_) = event_handler_pair { eprintln!("Failed to create event reader/writer pair session: {:?}", websocket_session); return Err(Error::ConnectionClosed) // TODO: Use proper error for this case to close the connection @@ -124,7 +130,7 @@ async fn serve_websocket(websocket_session: WebSocketSession, websocket: HyperWe continue; } let event_wrapper = events.unwrap(); - if event_wrapper.session_id != websocket_session_read_copy.session_id { + if event_wrapper.session_id != websocket_session_read_copy.session.hash { eprintln!("Websocket has session {:?} but was asked to write to session {} - skip.", websocket_session_read_copy, event_wrapper.session_id); continue; } @@ -150,6 +156,17 @@ async fn serve_websocket(websocket_session: WebSocketSession, websocket: HyperWe } else { println!("Received close message"); } + + let session_closed_event = SessionClosedDto { + session: websocket_session_read_copy.session.clone(), + reason: "ws closed".to_owned() + }; + let msg = json!(session_closed_event).to_string(); + let session_event_write_res = event_writer.write_to_session("session", &msg); + if let Err(e) = session_event_write_res { + eprintln!("Failed to write session closed event: {0}", e) + } + break; }, _ => {} } @@ -178,11 +195,18 @@ async fn serve_websocket(websocket_session: WebSocketSession, websocket: HyperWe println!("Sending kafka messages through websocket."); let send_res = websocket_writer.send(message).await; if let Err(e) = send_res { - eprintln!("Failed to send message to websocket for session {:?}: {:?}", websocket_session_write_copy, e) + eprintln!("Failed to send message to websocket for session {:?}: {:?}", websocket_session_write_copy, e); + match e { + tungstenite::error::Error::ConnectionClosed | tungstenite::error::Error::AlreadyClosed => { + println!("Websocket Connection for session {:?} is closed. Exiting kafka consumer.", websocket_session_write_copy); + break; + }, + _ => {} + } } // Avoid starvation of read thread (?) // TODO: How to avoid this? This is very bad for performance. - sleep(Duration::from_millis(10)).await; + sleep(Duration::from_millis(100)).await; } }); Ok(()) @@ -218,6 +242,7 @@ async fn handle_get_session(session_manager: &Arc>, req: R } async fn handle_session_create(session_manager: &Arc>, req: Request, addr: SocketAddr) -> Result, Infallible> { + println!("Called to create new session: {:?}", req); let mut locked = session_manager.lock().await; let player = Player {id: addr.to_string()}; let session_create_res = locked.create_session(player.clone()).await; @@ -355,7 +380,13 @@ struct SessionCreatedDto { player: Player, } -#[derive(Debug, Clone)] +#[derive(Debug, Serialize)] +struct SessionClosedDto { + session: Session, + reason: String +} + +#[derive(Debug, Clone, PartialEq)] enum WebSocketConnectionType { HOST, PEER, OBSERVER } @@ -376,7 +407,7 @@ impl FromStr for WebSocketConnectionType { #[derive(Debug, Clone)] struct WebSocketSession { pub connection_type: WebSocketConnectionType, - pub session_id: String + pub session: Session } impl WebSocketConnectionType { diff --git a/server/src/utils.rs b/server/src/utils.rs index 4bfdb7a..5b5cc13 100644 --- a/server/src/utils.rs +++ b/server/src/utils.rs @@ -56,3 +56,15 @@ pub mod http_utils_tests { assert_eq!(res, expected) } } + +pub mod time_utils { + use std::time::{SystemTime, UNIX_EPOCH}; + + pub fn now() -> u128 { + let start = SystemTime::now(); + let since_the_epoch = start + .duration_since(UNIX_EPOCH) + .expect("Time went backwards"); + return since_the_epoch.as_millis() + } +}