From b77ce89e80d94566d869bca022d4bed24ef3c810 Mon Sep 17 00:00:00 2001 From: Thilo Behnke Date: Sat, 11 Jun 2022 15:20:32 +0200 Subject: [PATCH] working input transmission --- pong/src/collision.rs | 45 +++-- pong/src/event.rs | 35 ++-- pong/src/game_field.rs | 187 +++++++++++-------- pong/src/game_object.rs | 6 +- pong/src/geom.rs | 2 +- pong/src/lib.rs | 2 +- pong/src/pong.rs | 32 ++-- pong/src/utils.rs | 8 +- pong/tests/collision_handler_test.rs | 2 +- pong/tests/collision_tests.rs | 8 +- pong/tests/game_object_tests.rs | 2 +- pong/tests/pong_tests.rs | 14 +- server/src/http.rs | 263 ++++++++++++++++++--------- server/src/kafka.rs | 125 +++++++------ server/src/main.rs | 11 +- server/src/player.rs | 4 +- server/src/session.rs | 152 +++++++++++----- server/src/utils.rs | 38 ++-- server/tests/utils_tests.rs | 2 +- src/lib.rs | 19 +- www/index.html | 45 ++++- www/index.js | 37 +++- 22 files changed, 671 insertions(+), 368 deletions(-) diff --git a/pong/src/collision.rs b/pong/src/collision.rs index 9d40872..7b5b73d 100644 --- a/pong/src/collision.rs +++ b/pong/src/collision.rs @@ -1,23 +1,21 @@ pub mod collision { use crate::game_object::game_object::GameObject; use crate::geom::geom::Vector; + use crate::utils::utils::{Logger, LoggerFactory}; use std::alloc::handle_alloc_error; use std::any::Any; use std::cell::{Ref, RefCell, RefMut}; use std::collections::HashMap; use std::fmt::Debug; use std::rc::Rc; - use crate::utils::utils::{Logger, LoggerFactory}; pub struct CollisionDetectorConfig { - groups: Vec + groups: Vec, } impl CollisionDetectorConfig { pub fn new() -> CollisionDetectorConfig { - CollisionDetectorConfig { - groups: vec![] - } + CollisionDetectorConfig { groups: vec![] } } pub fn matches_any_group(&self, type_a: &str, type_b: &str) -> bool { @@ -36,7 +34,7 @@ pub mod collision { pub struct CollisionDetector { config: CollisionDetectorConfig, - logger: Box + logger: Box, } impl CollisionDetector { @@ -44,7 +42,7 @@ pub mod collision { let logger = logger_factory.get("collision_detector"); CollisionDetector { config: CollisionDetectorConfig::new(), - logger + logger, } } @@ -68,7 +66,10 @@ pub mod collision { let rest = &objs[i..]; for other in rest.iter().map(|o| o.borrow()) { - if !self.config.matches_any_group(obj.obj_type(), other.obj_type()) { + if !self + .config + .matches_any_group(obj.obj_type(), other.obj_type()) + { // self.logger.log(&*format!("objs {} and {} do not match any group: {:?}", obj.obj_type(), other.obj_type(), self.config.groups)); continue; } @@ -116,14 +117,19 @@ pub mod collision { } pub struct CollisionHandlerRegistry { - handlers: HashMap<(String, String), fn(Rc>>, Rc>>)> + handlers: HashMap< + (String, String), + fn(Rc>>, Rc>>), + >, } type CollisionCallback = fn(Rc>>, Rc>>); impl CollisionHandlerRegistry { pub fn new() -> CollisionHandlerRegistry { - CollisionHandlerRegistry {handlers: HashMap::new()} + CollisionHandlerRegistry { + handlers: HashMap::new(), + } } pub fn add(&mut self, mapping: (String, String), callback: CollisionCallback) { @@ -136,7 +142,14 @@ pub mod collision { self.handlers.insert(mapping, callback); } - pub fn call(&self, mapping: &(String, String), values: (Rc>>, Rc>>)) -> bool { + pub fn call( + &self, + mapping: &(String, String), + values: ( + Rc>>, + Rc>>, + ), + ) -> bool { let regular = self.handlers.get(&mapping); if let Some(callback) = regular { callback(values.0, values.1); @@ -156,7 +169,7 @@ pub mod collision { pub struct CollisionHandler { logger: Box, - handlers: CollisionHandlerRegistry + handlers: CollisionHandlerRegistry, } impl CollisionHandler { @@ -181,10 +194,14 @@ pub mod collision { obj_a: Rc>>, obj_b: Rc>>, ) -> bool { - let key = (RefCell::borrow(&obj_a).obj_type().to_string(), RefCell::borrow(&obj_b).obj_type().to_string()); + let key = ( + RefCell::borrow(&obj_a).obj_type().to_string(), + RefCell::borrow(&obj_b).obj_type().to_string(), + ); let handler_res = self.handlers.call(&key, (obj_a, obj_b)); if !handler_res { - self.logger.log(&*format!("Found no matching collision handler: {:?}", key)); + self.logger + .log(&*format!("Found no matching collision handler: {:?}", key)); return false; } return true; diff --git a/pong/src/event.rs b/pong/src/event.rs index 48f8769..e29b11a 100644 --- a/pong/src/event.rs +++ b/pong/src/event.rs @@ -1,32 +1,35 @@ - pub mod event { + use serde::{Deserialize, Serialize}; use std::fmt::Debug; use std::fs::OpenOptions; use std::io::Write; - use serde::{Deserialize, Serialize}; #[derive(Debug, Deserialize, Serialize)] pub struct Event { pub topic: String, pub key: Option, - pub msg: String + pub msg: String, } - pub trait EventWriterImpl : Send + Sync { + pub trait EventWriterImpl: Send + Sync { fn write(&mut self, event: Event) -> Result<(), String>; } pub struct FileEventWriterImpl {} impl EventWriterImpl for FileEventWriterImpl { fn write(&mut self, event: Event) -> Result<(), String> { - let options = OpenOptions::new().read(true).create(true).write(true).open("events.log"); + let options = OpenOptions::new() + .read(true) + .create(true) + .write(true) + .open("events.log"); if let Err(e) = options { return Err(format!("{}", e)); } let mut file = options.unwrap(); match file.write(event.msg.as_bytes()) { Ok(_) => Ok(()), - Err(e) => Err(format!("{}", e)) + Err(e) => Err(format!("{}", e)), } } } @@ -39,46 +42,42 @@ pub mod event { } pub struct EventWriter { - writer_impl: Box + writer_impl: Box, } impl EventWriter { pub fn new(writer_impl: Box) -> EventWriter { - EventWriter { - writer_impl - } + EventWriter { writer_impl } } pub fn noop() -> EventWriter { EventWriter { - writer_impl: Box::new(NoopEventWriterImpl {}) + writer_impl: Box::new(NoopEventWriterImpl {}), } } pub fn file() -> EventWriter { EventWriter { - writer_impl: Box::new(FileEventWriterImpl {}) + writer_impl: Box::new(FileEventWriterImpl {}), } } pub fn write(&mut self, event: Event) -> Result<(), String> { - self.writer_impl.write(event) + self.writer_impl.write(event) } } - pub trait EventReaderImpl : Send + Sync { + pub trait EventReaderImpl: Send + Sync { fn read(&mut self) -> Result, String>; } pub struct EventReader { - reader_impl: Box + reader_impl: Box, } impl EventReader { pub fn new(reader_impl: Box) -> EventReader { - EventReader { - reader_impl - } + EventReader { reader_impl } } pub fn read(&mut self) -> Result, String> { diff --git a/pong/src/game_field.rs b/pong/src/game_field.rs index c6e99b4..23356a2 100644 --- a/pong/src/game_field.rs +++ b/pong/src/game_field.rs @@ -1,17 +1,23 @@ -use crate::collision::collision::{Collision, CollisionDetector, CollisionGroup, CollisionHandler, CollisionRegistry, Collisions}; +use crate::collision::collision::{ + Collision, CollisionDetector, CollisionGroup, CollisionHandler, CollisionRegistry, Collisions, +}; +use crate::event::event::Event; use crate::game_object::components::{DefaultGeomComp, DefaultPhysicsComp}; use crate::game_object::game_object::{DefaultGameObject, GameObject}; use crate::geom::geom::Vector; use crate::geom::shape::{Shape, ShapeType}; -use crate::pong::pong_collisions::{handle_ball_bounds_collision, handle_player_ball_collision, handle_player_bound_collision}; -use crate::pong::pong_events::{PongEventWriter, DefaultPongEventWriter, NoopPongEventWriter, PongEventType, GameObjUpdate}; +use crate::pong::pong_collisions::{ + handle_ball_bounds_collision, handle_player_ball_collision, handle_player_bound_collision, +}; +use crate::pong::pong_events::{ + DefaultPongEventWriter, GameObjUpdate, NoopPongEventWriter, PongEventType, PongEventWriter, +}; use crate::utils::utils::{DefaultLoggerFactory, Logger, LoggerFactory, NoopLogger}; use std::borrow::{Borrow, BorrowMut}; use std::cell::{Cell, Ref, RefCell, RefMut}; use std::collections::HashMap; use std::ops::Deref; use std::rc::Rc; -use crate::event::event::Event; #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub enum InputType { @@ -23,7 +29,7 @@ pub enum InputType { pub struct Input { pub input: InputType, pub obj_id: u16, - pub player: u16 + pub player: u16, } pub struct Field { @@ -39,7 +45,10 @@ pub struct Field { } impl Field { - pub fn new(logger_factory: Box, event_writer: Box) -> Field { + pub fn new( + logger_factory: Box, + event_writer: Box, + ) -> Field { let width = 800; let height = 600; @@ -55,7 +64,7 @@ impl Field { collision_detector: CollisionDetector::new(&logger_factory), collision_handler: CollisionHandler::new(&logger_factory), event_writer, - logger_factory + logger_factory, }; field.add_player(0, 0 + width / 20, height / 2); @@ -77,19 +86,17 @@ impl Field { handle_player_bound_collision, ); - field.collision_detector.set_groups( - vec![ - CollisionGroup(String::from("player"), String::from("ball")), - CollisionGroup(String::from("player"), String::from("bound")), - CollisionGroup(String::from("ball"), String::from("bound")), - ] - ); + field.collision_detector.set_groups(vec![ + CollisionGroup(String::from("player"), String::from("ball")), + CollisionGroup(String::from("player"), String::from("bound")), + CollisionGroup(String::from("ball"), String::from("bound")), + ]); return field; } pub fn mock(width: u16, height: u16) -> Field { - let logger_factory = DefaultLoggerFactory::new(Box::new(NoopLogger{})); + let logger_factory = DefaultLoggerFactory::new(Box::new(NoopLogger {})); let event_writer = NoopPongEventWriter::new(); Field { logger: logger_factory.get("game_field"), @@ -164,18 +171,36 @@ impl Field { let collision_handler = &self.collision_handler; let registered_collisions = collisions.get_collisions(); - self.logger.log(&*format!("Found {} collisions: {:?}", registered_collisions.len(), registered_collisions)); + self.logger.log(&*format!( + "Found {} collisions: {:?}", + registered_collisions.len(), + registered_collisions + )); for collision in registered_collisions.iter() { let objs = &self.objs; - let obj_a = objs.iter().find(|o| RefCell::borrow(o).id() == collision.0).unwrap().clone(); - let obj_b = objs.iter().find(|o| RefCell::borrow(o).id() == collision.1).unwrap().clone(); + let obj_a = objs + .iter() + .find(|o| RefCell::borrow(o).id() == collision.0) + .unwrap() + .clone(); + let obj_b = objs + .iter() + .find(|o| RefCell::borrow(o).id() == collision.1) + .unwrap() + .clone(); collision_handler.handle(obj_a, obj_b); } { for obj in self.objs.iter().filter(|o| RefCell::borrow(o).is_dirty()) { let mut obj = RefCell::borrow_mut(obj); - self.event_writer.write(PongEventType::GameObjUpdate(GameObjUpdate{obj_id: &obj.id().to_string(), vel: obj.vel(), orientation: obj.orientation(), pos: obj.pos()})); + self.event_writer + .write(PongEventType::GameObjUpdate(GameObjUpdate { + obj_id: &obj.id().to_string(), + vel: obj.vel(), + orientation: obj.orientation(), + pos: obj.pos(), + })); obj.set_dirty(false); } } @@ -231,62 +256,74 @@ impl DefaultGameObject { impl DefaultGameObject { pub fn bounds(width: u16, height: u16) -> Vec { let bounds = vec![ - Bounds(Bound::BOTTOM, Box::new(DefaultGameObject::new( - 90, - "bound".to_string(), - Box::new(DefaultGeomComp::new(Shape::rect( - Vector { - x: (width / 2) as f64, - y: 0 as f64, - }, - Vector::new(1., 0.), - width as f64, - 2., - ))), - Box::new(DefaultPhysicsComp::new_static()), - ))), - Bounds(Bound::TOP, Box::new(DefaultGameObject::new( - 91, - "bound".to_string(), - Box::new(DefaultGeomComp::new(Shape::rect( - Vector { - x: (width / 2) as f64, - y: height as f64, - }, - Vector::new(-1., 0.), - width as f64, - 2., - ))), - Box::new(DefaultPhysicsComp::new_static()), - ))), - Bounds(Bound::LEFT, Box::new(DefaultGameObject::new( - 92, - "bound".to_string(), - Box::new(DefaultGeomComp::new(Shape::rect( - Vector { - x: 0 as f64, - y: (height / 2) as f64, - }, - Vector::new(0., 1.), - 2., - height as f64, - ))), - Box::new(DefaultPhysicsComp::new_static()), - ))), - Bounds(Bound::RIGHT, Box::new(DefaultGameObject::new( - 93, - "bound".to_string(), - Box::new(DefaultGeomComp::new(Shape::rect( - Vector { - x: width as f64, - y: (height / 2) as f64, - }, - Vector::new(0., -1.), - 2., - height as f64, - ))), - Box::new(DefaultPhysicsComp::new_static()), - ))), + Bounds( + Bound::BOTTOM, + Box::new(DefaultGameObject::new( + 90, + "bound".to_string(), + Box::new(DefaultGeomComp::new(Shape::rect( + Vector { + x: (width / 2) as f64, + y: 0 as f64, + }, + Vector::new(1., 0.), + width as f64, + 2., + ))), + Box::new(DefaultPhysicsComp::new_static()), + )), + ), + Bounds( + Bound::TOP, + Box::new(DefaultGameObject::new( + 91, + "bound".to_string(), + Box::new(DefaultGeomComp::new(Shape::rect( + Vector { + x: (width / 2) as f64, + y: height as f64, + }, + Vector::new(-1., 0.), + width as f64, + 2., + ))), + Box::new(DefaultPhysicsComp::new_static()), + )), + ), + Bounds( + Bound::LEFT, + Box::new(DefaultGameObject::new( + 92, + "bound".to_string(), + Box::new(DefaultGeomComp::new(Shape::rect( + Vector { + x: 0 as f64, + y: (height / 2) as f64, + }, + Vector::new(0., 1.), + 2., + height as f64, + ))), + Box::new(DefaultPhysicsComp::new_static()), + )), + ), + Bounds( + Bound::RIGHT, + Box::new(DefaultGameObject::new( + 93, + "bound".to_string(), + Box::new(DefaultGeomComp::new(Shape::rect( + Vector { + x: width as f64, + y: (height / 2) as f64, + }, + Vector::new(0., -1.), + 2., + height as f64, + ))), + Box::new(DefaultPhysicsComp::new_static()), + )), + ), ]; bounds } diff --git a/pong/src/game_object.rs b/pong/src/game_object.rs index c37e00a..62100ca 100644 --- a/pong/src/game_object.rs +++ b/pong/src/game_object.rs @@ -28,7 +28,7 @@ pub mod game_object { pub obj_type: String, geom: Box, physics: Box, - dirty: bool + dirty: bool, } impl DefaultGameObject { @@ -43,7 +43,7 @@ pub mod game_object { obj_type, geom, physics, - dirty: false + dirty: false, } } } @@ -112,7 +112,7 @@ pub mod game_object { } fn is_dirty(&self) -> bool { - return self.dirty + return self.dirty; } fn set_dirty(&mut self, is_dirty: bool) { diff --git a/pong/src/geom.rs b/pong/src/geom.rs index 97bba43..895c022 100644 --- a/pong/src/geom.rs +++ b/pong/src/geom.rs @@ -1,5 +1,5 @@ pub mod geom { - use serde::{Serialize}; + use serde::Serialize; #[derive(Debug, Clone, Serialize)] pub struct Vector { diff --git a/pong/src/lib.rs b/pong/src/lib.rs index 0ba1216..baea5d0 100644 --- a/pong/src/lib.rs +++ b/pong/src/lib.rs @@ -1,7 +1,7 @@ pub mod collision; +pub mod event; pub mod game_field; pub mod game_object; pub mod geom; pub mod pong; pub mod utils; -pub mod event; diff --git a/pong/src/pong.rs b/pong/src/pong.rs index c273934..6be73e4 100644 --- a/pong/src/pong.rs +++ b/pong/src/pong.rs @@ -1,10 +1,10 @@ pub mod pong_collisions { - use std::cell::{RefCell, RefMut}; - use std::ops::Add; - use std::rc::Rc; use crate::game_object::game_object::GameObject; use crate::geom::geom::Vector; use crate::geom::shape::ShapeType; + use std::cell::{RefCell, RefMut}; + use std::ops::Add; + use std::rc::Rc; pub fn handle_player_ball_collision( ball: Rc>>, @@ -81,14 +81,14 @@ pub mod pong_collisions { } pub mod pong_events { - use serde_json::json; - use serde::{Serialize}; use crate::event::event::{Event, EventWriter}; use crate::geom::geom::Vector; + use serde::Serialize; + use serde_json::json; #[derive(Serialize)] pub enum PongEventType<'a> { - GameObjUpdate(GameObjUpdate<'a>) + GameObjUpdate(GameObjUpdate<'a>), } #[derive(Serialize)] @@ -96,7 +96,7 @@ pub mod pong_events { pub obj_id: &'a str, pub pos: &'a Vector, pub vel: &'a Vector, - pub orientation: &'a Vector + pub orientation: &'a Vector, } pub trait PongEventWriter { @@ -104,19 +104,17 @@ pub mod pong_events { } pub struct DefaultPongEventWriter { - writer: EventWriter + writer: EventWriter, } impl PongEventWriter for DefaultPongEventWriter { fn write(&mut self, event: PongEventType) -> Result<(), String> { let out_event = match event { - PongEventType::GameObjUpdate(ref update) => { - Event { - topic: String::from("obj_update"), - key: Some(update.obj_id.clone().to_string()), - msg: serde_json::to_string(&event).unwrap() - } - } + PongEventType::GameObjUpdate(ref update) => Event { + topic: String::from("obj_update"), + key: Some(update.obj_id.clone().to_string()), + msg: serde_json::to_string(&event).unwrap(), + }, }; self.writer.write(out_event) } @@ -126,7 +124,7 @@ pub mod pong_events { impl NoopPongEventWriter { pub fn new() -> Box { Box::new(DefaultPongEventWriter { - writer: EventWriter::noop() + writer: EventWriter::noop(), }) } } @@ -134,7 +132,7 @@ pub mod pong_events { impl DefaultPongEventWriter { pub fn new() -> Box { Box::new(DefaultPongEventWriter { - writer: EventWriter::file() + writer: EventWriter::file(), }) } } diff --git a/pong/src/utils.rs b/pong/src/utils.rs index c1f3b9a..fc996a6 100644 --- a/pong/src/utils.rs +++ b/pong/src/utils.rs @@ -4,7 +4,7 @@ pub mod utils { } pub struct DefaultLoggerFactory { - proto: Box + proto: Box, } impl LoggerFactory for DefaultLoggerFactory { @@ -17,10 +17,12 @@ pub mod utils { impl DefaultLoggerFactory { pub fn new(proto: Box) -> Box { - Box::new(DefaultLoggerFactory {proto}) + Box::new(DefaultLoggerFactory { proto }) } pub fn noop() -> Box { - Box::new(DefaultLoggerFactory {proto: Box::new(NoopLogger {})}) + Box::new(DefaultLoggerFactory { + proto: Box::new(NoopLogger {}), + }) } } diff --git a/pong/tests/collision_handler_test.rs b/pong/tests/collision_handler_test.rs index 22c5d71..325c45a 100644 --- a/pong/tests/collision_handler_test.rs +++ b/pong/tests/collision_handler_test.rs @@ -3,11 +3,11 @@ use pong::game_object::components::{DefaultGeomComp, DefaultPhysicsComp}; use pong::game_object::game_object::{DefaultGameObject, GameObject}; use pong::geom::geom::Vector; use pong::geom::shape::Shape; +use pong::utils::utils::DefaultLoggerFactory; use rstest::rstest; use std::borrow::{Borrow, BorrowMut}; use std::cell::RefCell; use std::rc::Rc; -use pong::utils::utils::DefaultLoggerFactory; #[rstest] #[case( diff --git a/pong/tests/collision_tests.rs b/pong/tests/collision_tests.rs index 7586031..40d8319 100644 --- a/pong/tests/collision_tests.rs +++ b/pong/tests/collision_tests.rs @@ -2,10 +2,10 @@ use pong::collision::collision::{Collision, CollisionDetector, CollisionGroup}; use pong::game_object::game_object::GameObject; use pong::geom::geom::{BoundingBox, Vector}; use pong::geom::shape::ShapeType; +use pong::utils::utils::DefaultLoggerFactory; use rstest::rstest; use std::cell::{Ref, RefCell}; use std::rc::Rc; -use pong::utils::utils::DefaultLoggerFactory; #[rstest] #[case(vec![], vec![])] @@ -67,7 +67,11 @@ pub struct MockGameObject { } impl MockGameObject { - pub fn new(id: u16, obj_type: &str, bounding_box: BoundingBox) -> Rc>> { + pub fn new( + id: u16, + obj_type: &str, + bounding_box: BoundingBox, + ) -> Rc>> { Rc::new(RefCell::new(Box::new(MockGameObject { id, obj_type: String::from(obj_type), diff --git a/pong/tests/game_object_tests.rs b/pong/tests/game_object_tests.rs index a50e4e1..8c36919 100644 --- a/pong/tests/game_object_tests.rs +++ b/pong/tests/game_object_tests.rs @@ -12,7 +12,7 @@ pub fn should_update_pos( #[case] start_pos: Vector, #[case] vel: Vector, #[case] expected_pos: Vector, - #[case] ms_diff: f64 + #[case] ms_diff: f64, ) { let mut obj = DefaultGameObject::new( 1, diff --git a/pong/tests/pong_tests.rs b/pong/tests/pong_tests.rs index b6bb2e4..268bd42 100644 --- a/pong/tests/pong_tests.rs +++ b/pong/tests/pong_tests.rs @@ -1,11 +1,11 @@ -use std::cell::RefCell; -use std::rc::Rc; -use rstest::rstest; use pong::game_field::{Bound, Field}; use pong::game_object::game_object::{DefaultGameObject, GameObject}; use pong::geom::geom::Vector; use pong::pong::pong_collisions::handle_player_bound_collision; use pong::utils::utils::{DefaultLoggerFactory, NoopLogger}; +use rstest::rstest; +use std::cell::RefCell; +use std::rc::Rc; #[rstest] #[case( @@ -44,7 +44,7 @@ pub fn should_correctly_handle_player_bounds_collision( #[case] mut player: Rc>>, #[case] mut bounds: Rc>>, #[case] mut player_expected: Rc>>, - #[case] mut bounds_expected: Rc>> + #[case] mut bounds_expected: Rc>>, ) { handle_player_bound_collision(player.clone(), bounds.clone()); assert_eq!(player_expected.borrow().pos(), player.borrow().pos()); @@ -65,7 +65,7 @@ fn get_bound(bound: Bound) -> Rc>> { let logger = DefaultLoggerFactory::noop(); let field = Field::new(logger); let bounds = DefaultGameObject::bounds(field.width, field.height); - return Rc::new(RefCell::new(bounds.into_iter().find(|b| { - b.0 == bound - }).unwrap().inner())); + return Rc::new(RefCell::new( + bounds.into_iter().find(|b| b.0 == bound).unwrap().inner(), + )); } diff --git a/server/src/http.rs b/server/src/http.rs index 7815825..d1997e9 100644 --- a/server/src/http.rs +++ b/server/src/http.rs @@ -1,3 +1,18 @@ +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; +use futures::{sink::SinkExt, stream::StreamExt}; +use hyper::server::conn::AddrStream; +use hyper::service::{make_service_fn, service_fn}; +use hyper::{body, Body, Method, Request, Response, Server, StatusCode}; +use hyper_tungstenite::tungstenite::{Error, Message}; +use hyper_tungstenite::{tungstenite, HyperWebsocket}; +use kafka::producer::Producer; +use pong::event::event::{Event, EventReader, EventWriter}; +use serde::{Deserialize, Serialize}; +use serde_json::json; use std::convert::Infallible; use std::fs::read; use std::io::ErrorKind::NotFound; @@ -5,38 +20,27 @@ use std::net::SocketAddr; use std::str::FromStr; use std::sync::Arc; 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, tungstenite}; -use hyper_tungstenite::tungstenite::{Error, Message}; -use kafka::producer::Producer; -use serde_json::json; -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::sync::Mutex; 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], port: u16, - session_manager: Arc> + session_manager: Arc>, } impl HttpServer { pub fn new(addr: [u8; 4], port: u16, kafka_host: &str) -> HttpServer { let session_manager = Arc::new(Mutex::new(SessionManager::new(kafka_host))); - HttpServer {addr, port, session_manager} + HttpServer { + addr, + port, + session_manager, + } } - pub async fn run(self) -> Result<(), Box> { + pub async fn run(self) -> Result<(), Box> { let make_svc = make_service_fn(|socket: &AddrStream| { let mut session_manager = Arc::clone(&self.session_manager); let addr = socket.remote_addr(); @@ -45,23 +49,34 @@ impl HttpServer { let mut session_manager = Arc::clone(&session_manager); async move { if hyper_tungstenite::is_upgrade_request(&req) { - println!("Received request to upgrade to websocket connection: {:?}", req); + println!( + "Received request to upgrade to websocket connection: {:?}", + req + ); let params = get_query_params(&req); println!("Ws request params: {:?}", params); if !params.contains_key("session_id") { eprintln!("Missing session id request param for websocket connection, don't upgrade connection to ws."); - return build_error_res("Missing request param: session_id", StatusCode::BAD_REQUEST); + return build_error_res( + "Missing request param: session_id", + StatusCode::BAD_REQUEST, + ); } if !params.contains_key("connection_type") { eprintln!("Missing connection type request param for websocket connection, don't upgrade connection to ws."); - let res = build_error_res("Missing request param: connection_type", StatusCode::BAD_REQUEST); + let res = build_error_res( + "Missing request param: connection_type", + StatusCode::BAD_REQUEST, + ); return res; } let session_id = params.get("session_id").unwrap(); let connection_type_raw = params.get("connection_type").unwrap(); - let connection_type = WebSocketConnectionType::from_str(connection_type_raw); + let connection_type = + WebSocketConnectionType::from_str(connection_type_raw); if let Err(_) = connection_type { - let error = format!("Invalid connection type: {}", connection_type_raw); + let error = + format!("Invalid connection type: {}", connection_type_raw); eprintln!("{}", error); return build_error_res(error.as_str(), StatusCode::BAD_REQUEST); } @@ -72,20 +87,27 @@ impl HttpServer { return build_error_res(error.as_str(), StatusCode::NOT_FOUND); } let session = session.unwrap(); - let websocket_session = WebSocketSession {session: session.clone(), connection_type: connection_type.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(); + let (response, websocket) = + hyper_tungstenite::upgrade(req, None).unwrap(); // Spawn a task to handle the websocket connection. tokio::spawn(async move { - if let Err(e) = serve_websocket(websocket_session, websocket, session_manager).await { + if let Err(e) = + serve_websocket(websocket_session, websocket, session_manager) + .await + { eprintln!("Error in websocket connection: {:?}", e); } }); // Return the response so the spawned future can continue. - return Ok(response) + return Ok(response); } return handle_request(&session_manager, req, addr).await; @@ -104,22 +126,34 @@ impl HttpServer { } /// Handle a websocket connection. -async fn serve_websocket(websocket_session: WebSocketSession, websocket: HyperWebsocket, session_manager: Arc>) -> Result<(), Error> { +async fn serve_websocket( + websocket_session: WebSocketSession, + websocket: HyperWebsocket, + session_manager: Arc>, +) -> Result<(), Error> { let mut websocket = websocket.await?; let (mut websocket_writer, mut websocket_reader) = websocket.split(); let session_manager = session_manager.lock().await; - 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()); + 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 + 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 } let (mut event_reader, mut event_writer) = event_handler_pair.unwrap(); let websocket_session_read_copy = websocket_session.clone(); tokio::spawn(async move { - println!("Ready to read messages from ws connection: {:?}", websocket_session_read_copy); + println!( + "Ready to read messages from ws connection: {:?}", + websocket_session_read_copy + ); while let Some(message) = websocket_reader.next().await { match message.unwrap() { Message::Text(msg) => { @@ -144,22 +178,31 @@ async fn serve_websocket(websocket_session: WebSocketSession, websocket: HyperWe } } if any_error { - eprintln!("Failed to write at least one message for session {}", event_wrapper.session_id); + eprintln!( + "Failed to write at least one message for session {}", + event_wrapper.session_id + ); } else { - println!("Successfully wrote {} messages to kafka for session {:?}", event_count, websocket_session_read_copy) + println!( + "Successfully wrote {} messages to kafka for session {:?}", + event_count, websocket_session_read_copy + ) } - }, + } Message::Close(msg) => { // No need to send a reply: tungstenite takes care of this for you. if let Some(msg) = &msg { - println!("Received close message with code {} and message: {}", msg.code, msg.reason); + println!( + "Received close message with code {} and message: {}", + msg.code, msg.reason + ); } else { println!("Received close message"); } let session_closed_event = SessionClosedDto { session: websocket_session_read_copy.session.clone(), - reason: "ws closed".to_owned() + 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); @@ -167,7 +210,7 @@ async fn serve_websocket(websocket_session: WebSocketSession, websocket: HyperWe eprintln!("Failed to write session closed event: {0}", e) } break; - }, + } _ => {} } } @@ -175,38 +218,41 @@ async fn serve_websocket(websocket_session: WebSocketSession, websocket: HyperWe }); let websocket_session_write_copy = websocket_session.clone(); tokio::spawn(async move { - println!("Ready to read messages from kafka: {:?}", websocket_session_write_copy); + println!( + "Ready to read messages from kafka: {:?}", + websocket_session_write_copy + ); loop { println!("Reading messages from kafka."); let messages = event_reader.read_from_session(); if let Err(_) = messages { - eprintln!("Failed to read messages from kafka for session: {:?}", websocket_session_write_copy); + eprintln!( + "Failed to read messages from kafka for session: {:?}", + websocket_session_write_copy + ); continue; } // println!("Read messages for websocket_session {:?} from consumer: {:?}", websocket_session_write_copy, messages); let messages = messages.unwrap(); if messages.len() == 0 { println!("No new messages from kafka."); - continue; - } - println!("{} new messages from kafka.", messages.len()); - let json = serde_json::to_string(&messages).unwrap(); - let message = Message::from(json); - 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); - 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; - }, - _ => {} + } else { + println!("{} new messages from kafka.", messages.len()); + let json = serde_json::to_string(&messages).unwrap(); + let message = Message::from(json); + 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 + ); + break; } } // Avoid starvation of read thread (?) // TODO: How to avoid this? This is very bad for performance. - sleep(Duration::from_millis(100)).await; + sleep(Duration::from_millis(1)).await; } }); Ok(()) @@ -214,19 +260,28 @@ async fn serve_websocket(websocket_session: WebSocketSession, websocket: HyperWe // TODO: How to handle event writes/reads? This must be a websocket, but how to implement in hyper (if possible)? // https://github.com/de-vri-es/hyper-tungstenite-rs -async fn handle_request(session_manager: &Arc>, req: Request, addr: SocketAddr) -> Result, Infallible> { +async fn handle_request( + session_manager: &Arc>, + req: Request, + addr: SocketAddr, +) -> Result, Infallible> { println!("req to {} with method {}", req.uri().path(), req.method()); match (req.method(), req.uri().path()) { (&Method::GET, "/session") => handle_get_session(session_manager, req).await, - (&Method::POST, "/create_session") => handle_session_create(session_manager, req, addr).await, + (&Method::POST, "/create_session") => { + handle_session_create(session_manager, req, addr).await + } (&Method::POST, "/join_session") => handle_session_join(session_manager, req, addr).await, (&Method::POST, "/write") => handle_event_write(session_manager, req).await, (&Method::POST, "/read") => handle_event_read(session_manager, req).await, - _ => Ok(Response::new("unknown".into())) + _ => Ok(Response::new("unknown".into())), } } -async fn handle_get_session(session_manager: &Arc>, req: Request) -> Result, Infallible> { +async fn handle_get_session( + session_manager: &Arc>, + req: Request, +) -> Result, Infallible> { let mut locked = session_manager.lock().await; let query_params = get_query_params(&req); let session_id = query_params.get("session_id"); @@ -241,37 +296,61 @@ async fn handle_get_session(session_manager: &Arc>, req: R return build_success_res(&serde_json::to_string(&session.unwrap()).unwrap()); } -async fn handle_session_create(session_manager: &Arc>, req: Request, addr: SocketAddr) -> Result, Infallible> { +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 player = Player { + id: addr.to_string(), + }; let session_create_res = locked.create_session(player.clone()).await; if let Err(e) = session_create_res { - return Ok(Response::builder().status(StatusCode::INTERNAL_SERVER_ERROR).body(Body::from(e)).unwrap()); + return Ok(Response::builder() + .status(StatusCode::INTERNAL_SERVER_ERROR) + .body(Body::from(e)) + .unwrap()); } - let session_created = SessionCreatedDto {session: session_create_res.unwrap(), player}; + let session_created = SessionCreatedDto { + session: session_create_res.unwrap(), + player, + }; let serialized = json!(session_created); return build_success_res(&serialized.to_string()); } -async fn handle_session_join(session_manager: &Arc>, mut req: Request, addr: SocketAddr) -> Result, Infallible> { +async fn handle_session_join( + session_manager: &Arc>, + mut req: Request, + addr: SocketAddr, +) -> Result, Infallible> { println!("Received request to join session: {:?}", req); let mut locked = session_manager.lock().await; let body = read_json_body::(&mut req).await; - let player = Player {id: addr.to_string()}; + let player = Player { + id: addr.to_string(), + }; let session_join_res = locked.join_session(body.session_id, player.clone()).await; if let Err(e) = session_join_res { eprintln!("Failed to join session: {:?}", e); - return Ok(Response::builder().status(StatusCode::INTERNAL_SERVER_ERROR).body(Body::from(e)).unwrap()); + return Ok(Response::builder() + .status(StatusCode::INTERNAL_SERVER_ERROR) + .body(Body::from(e)) + .unwrap()); } let session = session_join_res.unwrap(); println!("Successfully joined session: {:?}", session); - let session_joined = SessionJoinedDto {session, player}; + let session_joined = SessionJoinedDto { session, player }; let serialized = json!(session_joined); return build_success_res(&serialized.to_string()); } -async fn handle_event_write(session_manager: &Arc>, mut req: Request) -> Result, Infallible> { +async fn handle_event_write( + session_manager: &Arc>, + mut req: Request, +) -> Result, Infallible> { let mut locked = session_manager.lock().await; let event = read_json_body::(&mut req).await; let writer = locked.get_session_writer(&event.session_id); @@ -296,10 +375,16 @@ async fn handle_event_write(session_manager: &Arc>, mut re build_success_res(&serde_json::to_string(&event).unwrap()) } -async fn handle_event_read(session_manager: &Arc>, mut req: Request) -> Result, Infallible> { +async fn handle_event_read( + session_manager: &Arc>, + mut req: Request, +) -> Result, Infallible> { let mut locked = session_manager.lock().await; let read_payload = read_json_body::(&mut req).await; - let reader = locked.get_session_reader(&read_payload.session_id, &["move", "status", "input", "session"]); + let reader = locked.get_session_reader( + &read_payload.session_id, + &["move", "status", "input", "session"], + ); if let Err(e) = reader { let err = format!("Failed to read events: {}", e); println!("{}", err); @@ -308,7 +393,10 @@ async fn handle_event_read(session_manager: &Arc>, mut req return Ok(res); } let mut reader = reader.unwrap(); - println!("Reading session events from kafka for session: {}", read_payload.session_id); + println!( + "Reading session events from kafka for session: {}", + read_payload.session_id + ); let events = reader.read_from_session(); if let Err(e) = events { let err = format!("Failed to read events: {}", e); @@ -327,7 +415,10 @@ pub fn build_success_res(value: &str) -> Result, Infallible> { let mut res = Response::new(Body::from(json)); let headers = res.headers_mut(); headers.insert("Content-Type", "application/json".parse().unwrap()); - headers.insert("Access-Control-Allow-Origin", "http://localhost:8080".parse().unwrap()); + headers.insert( + "Access-Control-Allow-Origin", + "http://localhost:8080".parse().unwrap(), + ); Ok(res) } @@ -348,24 +439,24 @@ async fn shutdown_signal() { #[derive(Debug, Deserialize, Serialize)] struct SessionEventListDTO { session_id: String, - events: Vec + events: Vec, } #[derive(Debug, Deserialize, Serialize)] struct SessionEventWriteDTO { session_id: String, topic: String, - msg: String + msg: String, } #[derive(Debug, Serialize, Deserialize)] struct SessionReadDTO { - session_id: String + session_id: String, } #[derive(Debug, Serialize, Deserialize)] struct SessionJoinDto { - session_id: String + session_id: String, } #[derive(Debug, Serialize)] @@ -383,12 +474,14 @@ struct SessionCreatedDto { #[derive(Debug, Serialize)] struct SessionClosedDto { session: Session, - reason: String + reason: String, } #[derive(Debug, Clone, PartialEq)] enum WebSocketConnectionType { - HOST, PEER, OBSERVER + HOST, + PEER, + OBSERVER, } impl FromStr for WebSocketConnectionType { @@ -399,7 +492,7 @@ impl FromStr for WebSocketConnectionType { "host" => Ok(WebSocketConnectionType::HOST), "peer" => Ok(WebSocketConnectionType::PEER), "observer" => Ok(WebSocketConnectionType::OBSERVER), - _ => Err(()) + _ => Err(()), } } } @@ -407,14 +500,16 @@ impl FromStr for WebSocketConnectionType { #[derive(Debug, Clone)] struct WebSocketSession { pub connection_type: WebSocketConnectionType, - pub session: Session + pub session: Session, } impl WebSocketConnectionType { pub fn get_topics(&self) -> &[&str] { match self { WebSocketConnectionType::HOST => &["input", "session"], - WebSocketConnectionType::PEER | WebSocketConnectionType::OBSERVER => &["move", "input", "status", "session"], + WebSocketConnectionType::PEER | WebSocketConnectionType::OBSERVER => { + &["move", "input", "status", "session"] + } } } } diff --git a/server/src/kafka.rs b/server/src/kafka.rs index 2185be2..9ffcbc3 100644 --- a/server/src/kafka.rs +++ b/server/src/kafka.rs @@ -1,20 +1,20 @@ +use crate::hash::Hasher; +use crate::session::Session; +use hyper::{Body, Client, Method, Request, Uri}; +use kafka::client::metadata::Topic; +use kafka::client::{KafkaClient, ProduceMessage}; +use kafka::consumer::{Consumer, FetchOffset, GroupOffsetStorage, MessageSet}; +use kafka::producer::{DefaultPartitioner, Partitioner, Producer, Record, RequiredAcks, Topics}; +use pong::event::event::{Event, EventReaderImpl, EventWriter, EventWriterImpl}; +use serde::Deserialize; use std::hash::{BuildHasher, Hash}; use std::process::ExitStatus; use std::str::FromStr; use std::time::Duration; -use hyper::{Body, Client, Method, Request, Uri}; -use serde::{Deserialize}; -use kafka::client::{KafkaClient, ProduceMessage}; -use kafka::client::metadata::Topic; use tokio::process::Command; -use kafka::consumer::{Consumer, FetchOffset, GroupOffsetStorage, MessageSet}; -use kafka::producer::{DefaultPartitioner, Partitioner, Producer, Record, RequiredAcks, Topics}; -use pong::event::event::{Event, EventReaderImpl, EventWriter, EventWriterImpl}; -use crate::hash::Hasher; -use crate::session::Session; pub struct KafkaSessionEventWriterImpl { - producer: Producer + producer: Producer, } impl KafkaSessionEventWriterImpl { pub fn new(host: &str) -> KafkaSessionEventWriterImpl { @@ -25,14 +25,12 @@ impl KafkaSessionEventWriterImpl { .with_partitioner(SessionPartitioner {}) .create() .unwrap(); - KafkaSessionEventWriterImpl { - producer - } + KafkaSessionEventWriterImpl { producer } } } pub struct KafkaDefaultEventWriterImpl { - producer: Producer + producer: Producer, } impl KafkaDefaultEventWriterImpl { pub fn new(host: &str) -> KafkaDefaultEventWriterImpl { @@ -42,13 +40,10 @@ impl KafkaDefaultEventWriterImpl { .with_required_acks(RequiredAcks::One) .create() .unwrap(); - KafkaDefaultEventWriterImpl { - producer - } + KafkaDefaultEventWriterImpl { producer } } } - impl EventWriterImpl for KafkaSessionEventWriterImpl { fn write(&mut self, event: Event) -> Result<(), String> { match event.key { @@ -56,14 +51,14 @@ impl EventWriterImpl for KafkaSessionEventWriterImpl { let record = Record::from_key_value(event.topic.as_str(), key, event.msg.as_str()); match self.producer.send(&record) { Ok(()) => Ok(()), - Err(e) => Err(format!("{}", e)) + Err(e) => Err(format!("{}", e)), } - }, + } None => { let record = Record::from_value(event.topic.as_str(), event.msg.as_str()); match self.producer.send(&record) { Ok(()) => Ok(()), - Err(e) => Err(format!("{}", e)) + Err(e) => Err(format!("{}", e)), } } } @@ -77,14 +72,14 @@ impl EventWriterImpl for KafkaDefaultEventWriterImpl { let record = Record::from_key_value(event.topic.as_str(), key, event.msg.as_str()); match self.producer.send(&record) { Ok(()) => Ok(()), - Err(e) => Err(format!("{}", e)) + Err(e) => Err(format!("{}", e)), } - }, + } None => { let record = Record::from_value(event.topic.as_str(), event.msg.as_str()); match self.producer.send(&record) { Ok(()) => Ok(()), - Err(e) => Err(format!("{}", e)) + Err(e) => Err(format!("{}", e)), } } } @@ -92,7 +87,7 @@ impl EventWriterImpl for KafkaDefaultEventWriterImpl { } pub struct KafkaEventReaderImpl { - consumer: Consumer + consumer: Consumer, } impl KafkaEventReaderImpl { pub fn default() -> KafkaEventReaderImpl { @@ -105,7 +100,7 @@ impl KafkaEventReaderImpl { pub fn new(host: &str) -> KafkaEventReaderImpl { println!("Connecting consumer to kafka host: {}", host); - let mut consumer = Consumer::from_hosts(vec!(host.to_owned())) + let mut consumer = Consumer::from_hosts(vec![host.to_owned()]) .with_topic("move".to_owned()) .with_topic("status".to_owned()) .with_topic("input".to_owned()) @@ -114,14 +109,16 @@ impl KafkaEventReaderImpl { .with_offset_storage(GroupOffsetStorage::Kafka) .create() .unwrap(); - KafkaEventReaderImpl { - consumer - } + KafkaEventReaderImpl { consumer } } - pub fn for_partitions(host: &str, partitions: &[i32], topics: &[&str]) -> Result { + pub fn for_partitions( + host: &str, + partitions: &[i32], + topics: &[&str], + ) -> Result { println!("Connecting partition specific consumer to kafka host {} with topics {:?} / partitions {:?}", host, topics, partitions); - let mut builder = Consumer::from_hosts(vec!(host.to_owned())); + let mut builder = Consumer::from_hosts(vec![host.to_owned()]); for topic in topics.iter() { builder = builder.with_topic_partitions(topic.parse().unwrap(), partitions); } @@ -130,16 +127,13 @@ impl KafkaEventReaderImpl { .with_group("group".to_owned()) .with_offset_storage(GroupOffsetStorage::Kafka); - let consumer = builder - .create(); + let consumer = builder.create(); if let Err(e) = consumer { eprintln!("Failed to connect consumer: {:?}", e); return Err("Failed to connect consumer".to_string()); } let consumer = consumer.unwrap(); - Ok(KafkaEventReaderImpl { - consumer - }) + Ok(KafkaEventReaderImpl { consumer }) } } impl EventReaderImpl for KafkaEventReaderImpl { @@ -162,11 +156,18 @@ impl KafkaEventReaderImpl { let partition = ms.partition(); println!("querying topic={} partition={}", topic, partition); for m in ms.messages() { - let event = Event {topic: String::from(topic), key: Some(std::str::from_utf8(m.key).unwrap().parse().unwrap()), msg: std::str::from_utf8(m.value).unwrap().parse().unwrap() }; + let event = Event { + topic: String::from(topic), + key: Some(std::str::from_utf8(m.key).unwrap().parse().unwrap()), + msg: std::str::from_utf8(m.value).unwrap().parse().unwrap(), + }; topic_event_count += 1; events.push(event); } - println!("returned {:?} events for topic={} partition={}", topic_event_count, topic, partition); + println!( + "returned {:?} events for topic={} partition={}", + topic_event_count, topic, partition + ); self.consumer.consume_messageset(ms).unwrap(); } self.consumer.commit_consumed().unwrap(); @@ -175,20 +176,22 @@ impl KafkaEventReaderImpl { } pub struct KafkaSessionEventReaderImpl { - inner: KafkaEventReaderImpl + inner: KafkaEventReaderImpl, } impl KafkaSessionEventReaderImpl { - pub fn new(host: &str, session: &Session, topics: &[&str]) -> Result { + pub fn new( + host: &str, + session: &Session, + topics: &[&str], + ) -> Result { let partitions = [session.id as i32]; let reader = KafkaEventReaderImpl::for_partitions(host, &partitions, topics); if let Err(e) = reader { return Err("Failed to create kafka session event reader".to_string()); } let reader = reader.unwrap(); - Ok(KafkaSessionEventReaderImpl { - inner: reader - }) + Ok(KafkaSessionEventReaderImpl { inner: reader }) } } @@ -198,24 +201,31 @@ impl EventReaderImpl for KafkaSessionEventReaderImpl { } } - #[derive(Debug)] pub struct KafkaTopicManager { - partition_management_endpoint: String + partition_management_endpoint: String, } impl KafkaTopicManager { - pub fn default() -> KafkaTopicManager { - KafkaTopicManager {partition_management_endpoint: "http://localhost:7243/add_partition".to_owned()} + KafkaTopicManager { + partition_management_endpoint: "http://localhost:7243/add_partition".to_owned(), + } } pub fn from(topic_manager_host: &str) -> KafkaTopicManager { - KafkaTopicManager {partition_management_endpoint: format!("http://{}/add_partition", topic_manager_host).to_owned()} + KafkaTopicManager { + partition_management_endpoint: format!("http://{}/add_partition", topic_manager_host) + .to_owned(), + } } pub async fn add_partition(&self) -> Result { let mut client = Client::new(); - let request = Request::builder().method(Method::POST).uri(Uri::from_str(&self.partition_management_endpoint).unwrap()).body(Body::empty()).unwrap(); + let request = Request::builder() + .method(Method::POST) + .uri(Uri::from_str(&self.partition_management_endpoint).unwrap()) + .body(Body::empty()) + .unwrap(); let res = client.request(request).await; if let Err(e) = res { let error = format!("Failed to add partition: {:?}", e); @@ -243,19 +253,26 @@ impl KafkaTopicManager { } let json = serde_json::from_str::(res_str.unwrap()); if let Err(e) = json { - let error = format!("Failed to convert string {} to json: {:?}", res_str.unwrap(), e); + let error = format!( + "Failed to convert string {} to json: {:?}", + res_str.unwrap(), + e + ); println!("{}", error); return Err(error); } let updated_partition_count = json.unwrap().data; - println!("Successfully created partition: {}", updated_partition_count); + println!( + "Successfully created partition: {}", + updated_partition_count + ); Ok(updated_partition_count) } } #[derive(Deserialize)] struct PartitionApiDTO { - data: u16 + data: u16, } pub struct SessionPartitioner {} @@ -267,8 +284,8 @@ impl Partitioner for SessionPartitioner { let key = std::str::from_utf8(key).unwrap(); msg.partition = key.parse::().unwrap(); // println!("Overriding message partition with key: {}", msg.partition); - }, - None => panic!("Producing message without key not allowed!") + } + None => panic!("Producing message without key not allowed!"), } } } diff --git a/server/src/main.rs b/server/src/main.rs index e4fb33c..c489265 100644 --- a/server/src/main.rs +++ b/server/src/main.rs @@ -2,14 +2,17 @@ extern crate core; use crate::http::HttpServer; +mod hash; pub mod http; pub mod kafka; -pub mod utils; -mod hash; -mod session; mod player; +mod session; +pub mod utils; #[tokio::main] pub async fn main() { - HttpServer::new([127, 0, 0, 1], 4000, "localhost:9093").run().await.expect("failed to run server"); + HttpServer::new([127, 0, 0, 1], 4000, "localhost:9093") + .run() + .await + .expect("failed to run server"); } diff --git a/server/src/player.rs b/server/src/player.rs index 2fce89f..771f050 100644 --- a/server/src/player.rs +++ b/server/src/player.rs @@ -1,7 +1,7 @@ use hyper::{Body, Request}; use serde::{Deserialize, Serialize}; -#[derive(Debug, Clone, Serialize, Deserialize)] +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] pub struct Player { - pub id: String + pub id: String, } diff --git a/server/src/session.rs b/server/src/session.rs index 0d36614..5e174fd 100644 --- a/server/src/session.rs +++ b/server/src/session.rs @@ -1,21 +1,24 @@ +use crate::hash::Hasher; +use crate::kafka::{ + KafkaDefaultEventWriterImpl, KafkaEventReaderImpl, KafkaSessionEventReaderImpl, + KafkaSessionEventWriterImpl, KafkaTopicManager, +}; +use crate::player::Player; +use kafka::producer::Producer; +use pong::event::event::{Event, EventReader, EventWriter}; +use serde::de::DeserializeOwned; +use serde::{Deserialize, Serialize}; +use serde_json::json; use std::collections::HashMap; use std::rc::Rc; use std::sync::Arc; -use kafka::producer::Producer; -use crate::hash::Hasher; -use crate::kafka::{KafkaDefaultEventWriterImpl, KafkaEventReaderImpl, KafkaSessionEventReaderImpl, KafkaSessionEventWriterImpl, KafkaTopicManager}; -use serde::{Serialize, Deserialize}; -use serde::de::DeserializeOwned; -use serde_json::json; use tokio::sync::Mutex; -use pong::event::event::{Event, EventReader, EventWriter}; -use crate::player::Player; pub struct SessionManager { kafka_host: String, sessions: Vec, session_producer: EventWriter, - topic_manager: KafkaTopicManager + topic_manager: KafkaTopicManager, } // TODO: On startup read the session events from kafka to restore the session id <-> hash mappings. @@ -25,12 +28,17 @@ impl SessionManager { kafka_host: kafka_host.to_owned(), sessions: vec![], topic_manager: KafkaTopicManager::from("localhost:7243"), - session_producer: EventWriter::new(Box::new(KafkaDefaultEventWriterImpl::new(kafka_host))) + session_producer: EventWriter::new(Box::new(KafkaDefaultEventWriterImpl::new( + kafka_host, + ))), } } pub fn get_session(&self, session_id: &str) -> Option { - self.sessions.iter().find(|s| s.hash == session_id).map_or_else(|| None, |s| Some(s.clone())) + self.sessions + .iter() + .find(|s| s.hash == session_id) + .map_or_else(|| None, |s| Some(s.clone())) } pub async fn create_session(&mut self, player: Player) -> Result { @@ -41,14 +49,18 @@ impl SessionManager { } let session_id = add_partition_res.unwrap(); let session_hash = Hasher::hash(session_id); - let session = Session::new (session_id, session_hash, player.clone()); + let session = Session::new(session_id, session_hash, player.clone()); println!("Successfully created session: {:?}", session); self.write_to_producer(session_created(session.clone(), player.clone())); self.sessions.push(session.clone()); Ok(session) } - pub async fn join_session(&mut self, session_id: String, player: Player) -> Result { + pub async fn join_session( + &mut self, + session_id: String, + player: Player, + ) -> Result { let updated_session = { let session = self.sessions.iter_mut().find(|s| s.hash == session_id); if let None = session { @@ -64,6 +76,10 @@ impl SessionManager { let error = format!("Can't join session with more than 1 player: {}", session_id); return Err(error); } + if session.players[0] == player { + let error = format!("Can't join session, because player {:?} is already in session: {}", player, session_id); + return Err(error); + } session.players.push(player.clone()); session.state = SessionState::RUNNING; session.clone() @@ -75,9 +91,16 @@ impl SessionManager { Ok(updated_session.clone()) } - fn write_to_producer(&mut self, session_event: T) -> Result<(), String> where T : Serialize { + fn write_to_producer(&mut self, session_event: T) -> Result<(), String> + where + T: Serialize, + { let json_event = serde_json::to_string(&session_event).unwrap(); - let session_event_write = self.session_producer.write(Event {topic: "session".to_owned(), key: None, msg: json_event}); + let session_event_write = self.session_producer.write(Event { + topic: "session".to_owned(), + key: None, + msg: json_event, + }); if let Err(e) = session_event_write { let message = format!("Failed to write session create event to kafka: {:?}", e); println!("{}", e); @@ -87,47 +110,65 @@ impl SessionManager { return Ok(()); } - pub fn split(&self, session_id: &str, read_topics: &[&str]) -> Result<(SessionReader, SessionWriter), String> { + pub fn split( + &self, + session_id: &str, + read_topics: &[&str], + ) -> Result<(SessionReader, SessionWriter), String> { let reader = self.get_session_reader(session_id, read_topics); if let Err(e) = reader { println!("Failed to create session reader: {:?}", e); - return Err("Failed to create session reader".to_string()) + return Err("Failed to create session reader".to_string()); } let writer = self.get_session_writer(session_id); if let Err(e) = writer { println!("Failed to create session writer: {:?}", e); - return Err("Failed to create session writer".to_string()) + return Err("Failed to create session writer".to_string()); } - return Ok((reader.unwrap(), writer.unwrap())) + return Ok((reader.unwrap(), writer.unwrap())); } - pub fn get_session_reader(&self, session_id: &str, topics: &[&str]) -> Result { + pub fn get_session_reader( + &self, + session_id: &str, + topics: &[&str], + ) -> Result { let session = self.find_session(&session_id); if let None = session { - return Err(format!("Unable to find session with hash {}", session_id)) + return Err(format!("Unable to find session with hash {}", session_id)); } let session = session.unwrap(); let kafka_reader = KafkaSessionEventReaderImpl::new(&self.kafka_host, &session, topics); if let Err(_) = kafka_reader { - return Err("Unable to create kafka reader.".to_string()) + return Err("Unable to create kafka reader.".to_string()); } let kafka_reader = kafka_reader.unwrap(); let event_reader = EventReader::new(Box::new(kafka_reader)); - Ok(SessionReader {reader: event_reader, session}) + Ok(SessionReader { + reader: event_reader, + session, + }) } pub fn get_session_writer(&self, session_id: &str) -> Result { let session = self.find_session(&session_id); if let None = session { - return Err(format!("Unable to find session with hash {}", session_id)) + return Err(format!("Unable to find session with hash {}", session_id)); } let session = session.unwrap(); - let event_writer = EventWriter::new(Box::new(KafkaSessionEventWriterImpl::new(&self.kafka_host))); - Ok(SessionWriter {writer: event_writer, session}) + let event_writer = + EventWriter::new(Box::new(KafkaSessionEventWriterImpl::new(&self.kafka_host))); + Ok(SessionWriter { + writer: event_writer, + session, + }) } fn find_session(&self, session_id: &str) -> Option { - self.sessions.iter().find(|s| session_id == s.hash).map(|s| s.clone()) + self.sessions + .iter() + .find(|s| session_id == s.hash) + .map(|s| s.clone()) } } @@ -154,26 +195,39 @@ impl CachingSessionManager { self.inner.create_session(player).await } - pub async fn join_session(&mut self, session_id: String, player: Player) -> Result { + pub async fn join_session( + &mut self, + session_id: String, + player: Player, + ) -> Result { self.inner.join_session(session_id, player).await } - pub fn get_session_reader(&mut self, session_id: &str) -> Result>, String> { + pub fn get_session_reader( + &mut self, + session_id: &str, + ) -> Result>, String> { let cached = self.reader_cache.get(session_id); if let Some(reader) = cached { println!("Reusing existing reader for session: {:?}", session_id); return Ok(Arc::clone(reader)); } - let reader = self.inner.get_session_reader(session_id, &["move", "input", "status", "session"]); + let reader = self + .inner + .get_session_reader(session_id, &["move", "input", "status", "session"]); if let Err(e) = reader { return Err(e); } let reader = Arc::new(Mutex::new(reader.unwrap())); - self.reader_cache.insert(session_id.to_string(), Arc::clone(&reader)); + self.reader_cache + .insert(session_id.to_string(), Arc::clone(&reader)); return Ok(Arc::clone(&reader)); } - pub fn get_session_writer(&mut self, session_id: &str) -> Result>, String> { + pub fn get_session_writer( + &mut self, + session_id: &str, + ) -> Result>, String> { let cached = self.writer_cache.get(session_id); if let Some(writer) = cached { println!("Reusing existing writer for session: {:?}", session_id); @@ -184,7 +238,8 @@ impl CachingSessionManager { return Err(e); } let writer = Arc::new(Mutex::new(writer.unwrap())); - self.writer_cache.insert(session_id.to_string(), Arc::clone(&writer)); + self.writer_cache + .insert(session_id.to_string(), Arc::clone(&writer)); return Ok(Arc::clone(&writer)); } } @@ -194,16 +249,16 @@ pub struct Session { pub id: u16, pub hash: String, pub state: SessionState, - players: Vec + players: Vec, } impl Session { pub fn new(id: u16, hash: String, player: Player) -> Session { Session { - players: vec!(player), + players: vec![player], id, hash, - state: SessionState::PENDING + state: SessionState::PENDING, } } @@ -213,7 +268,7 @@ impl Session { pub fn join(&mut self, player: Player) -> bool { if !self.can_be_joined() { - return false + return false; } self.players.push(player); return true; @@ -224,24 +279,28 @@ impl Session { pub enum SessionState { PENDING, // 1 player is missing RUNNING, // game is playing - CLOSED // game is over + CLOSED, // game is over } pub struct SessionWriter { session: Session, - writer: EventWriter + writer: EventWriter, } impl SessionWriter { pub fn write_to_session(&mut self, topic: &str, msg: &str) -> Result<(), String> { - let event = Event {msg: msg.to_owned(), key: Some(self.session.id.to_string()), topic: topic.to_owned()}; + let event = Event { + msg: msg.to_owned(), + key: Some(self.session.id.to_string()), + topic: topic.to_owned(), + }; self.writer.write(event) } } pub struct SessionReader { session: Session, - reader: EventReader + reader: EventReader, } impl SessionReader { @@ -254,7 +313,7 @@ impl SessionReader { struct SessionCreatedEvent { event_type: SessionEventType, session: Session, - player: Player + player: Player, } impl SessionCreatedEvent { @@ -262,7 +321,7 @@ impl SessionCreatedEvent { SessionCreatedEvent { event_type: SessionEventType::CREATED, session, - player + player, } } } @@ -271,7 +330,7 @@ impl SessionCreatedEvent { struct SessionJoinedEvent { event_type: SessionEventType, session: Session, - player: Player + player: Player, } impl SessionJoinedEvent { @@ -279,7 +338,7 @@ impl SessionJoinedEvent { SessionJoinedEvent { event_type: SessionEventType::JOINED, session, - player + player, } } } @@ -294,5 +353,6 @@ fn session_joined(session: Session, player: Player) -> SessionJoinedEvent { #[derive(Deserialize, Serialize)] enum SessionEventType { - CREATED, JOINED + CREATED, + JOINED, } diff --git a/server/src/utils.rs b/server/src/utils.rs index 5b5cc13..1ea3a3a 100644 --- a/server/src/utils.rs +++ b/server/src/utils.rs @@ -1,11 +1,11 @@ pub mod http_utils { + use hyper::body::Buf; + use hyper::{body, Body, Request}; + use serde::de::DeserializeOwned; + use serde::Deserialize; use std::borrow::BorrowMut; use std::collections::HashMap; use std::io::Read; - use hyper::{Body, body, Request}; - use hyper::body::Buf; - use serde::de::DeserializeOwned; - use serde::Deserialize; pub fn get_query_params(req: &Request) -> HashMap<&str, &str> { let uri = req.uri(); @@ -13,13 +13,18 @@ pub mod http_utils { println!("uri={:?}, query={:?}", uri, query); match query { None => HashMap::new(), - Some(query) => { - query.split("&").map(|s| s.split_at(s.find("=").unwrap())).map(|(key, value)| (key, &value[1..])).collect() - } + Some(query) => query + .split("&") + .map(|s| s.split_at(s.find("=").unwrap())) + .map(|(key, value)| (key, &value[1..])) + .collect(), } } - pub async fn read_json_body(req: &mut Request) -> T where T : DeserializeOwned { + pub async fn read_json_body(req: &mut Request) -> T + where + T: DeserializeOwned, + { let mut body = req.body_mut(); let bytes = body::to_bytes(body).await.unwrap(); let body_str = std::str::from_utf8(&*bytes).unwrap(); @@ -29,12 +34,12 @@ pub mod http_utils { #[cfg(test)] pub mod http_utils_tests { + use super::*; + use crate::utils::http_utils::get_query_params; + use hyper::http::uri::{Builder, Parts}; + use hyper::{Body, Request, Uri}; use rstest::rstest; use std::collections::HashMap; - use hyper::{Body, Request, Uri}; - use hyper::http::uri::{Builder, Parts}; - use crate::utils::http_utils::get_query_params; - use super::*; #[rstest] #[case( @@ -50,7 +55,12 @@ pub mod http_utils_tests { HashMap::from([("topic", "status"), ("key", "abc")]) )] fn get_query_params_tests(#[case] query_str: &str, #[case] expected: HashMap<&str, &str>) { - let uri = Builder::new().scheme("https").authority("behnke.rs").path_and_query(query_str).build().unwrap(); + let uri = Builder::new() + .scheme("https") + .authority("behnke.rs") + .path_and_query(query_str) + .build() + .unwrap(); let req = Request::get(uri).body(Body::empty()).unwrap(); let res = get_query_params(&req); assert_eq!(res, expected) @@ -65,6 +75,6 @@ pub mod time_utils { let since_the_epoch = start .duration_since(UNIX_EPOCH) .expect("Time went backwards"); - return since_the_epoch.as_millis() + return since_the_epoch.as_millis(); } } diff --git a/server/tests/utils_tests.rs b/server/tests/utils_tests.rs index 7eef968..afa3d9a 100644 --- a/server/tests/utils_tests.rs +++ b/server/tests/utils_tests.rs @@ -1,6 +1,6 @@ pub mod tests { - use std::collections::HashMap; use rstest::rstest; + use std::collections::HashMap; #[rstest] #[case( diff --git a/src/lib.rs b/src/lib.rs index a4f6aef..facd64e 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,18 +1,18 @@ mod utils; -use std::cell::RefCell; use pong::collision::collision::{Collision, CollisionDetector}; use pong::game_field::{Field, Input, InputType}; use pong::game_object::game_object::{DefaultGameObject, GameObject}; use pong::geom::geom::Vector; use pong::geom::shape::ShapeType; +use pong::pong::pong_events::DefaultPongEventWriter; use pong::utils::utils::{DefaultLoggerFactory, Logger}; use serde::{Deserialize, Serialize}; use serde_json::json; +use std::cell::RefCell; use std::cmp::{max, min}; use std::rc::Rc; use wasm_bindgen::prelude::*; -use pong::pong::pong_events::DefaultPongEventWriter; extern crate serde_json; extern crate web_sys; @@ -93,7 +93,7 @@ impl InputTypeDTO { pub struct InputDTO { pub input: InputTypeDTO, pub obj_id: u16, - pub player: u16 + pub player: u16, } impl InputDTO { @@ -101,7 +101,7 @@ impl InputDTO { return Input { input: self.input.to_input_type(), obj_id: self.obj_id, - player: self.player + player: self.player, }; } } @@ -114,7 +114,10 @@ pub struct FieldWrapper { #[wasm_bindgen] impl FieldWrapper { pub fn new() -> FieldWrapper { - let field = Field::new(DefaultLoggerFactory::new(Box::new(WasmLogger::root())), DefaultPongEventWriter::new()); + let field = Field::new( + DefaultLoggerFactory::new(Box::new(WasmLogger::root())), + DefaultPongEventWriter::new(), + ); FieldWrapper { field } } @@ -151,12 +154,14 @@ impl FieldWrapper { #[derive(Clone)] pub struct WasmLogger { - name: String + name: String, } impl WasmLogger { pub fn root() -> WasmLogger { - WasmLogger {name: String::from("root")} + WasmLogger { + name: String::from("root"), + } } } diff --git a/www/index.html b/www/index.html index 291697c..bce07fb 100644 --- a/www/index.html +++ b/www/index.html @@ -16,12 +16,27 @@ justify-content: center; overflow: hidden; } + + .game_input { + border: 1px solid grey; + border-radius: 0.5em; + text-align: center; + } + + .game_input--inactive { + color: black; + } + + .game_input--active { + color: white; + background-color: red; + } -
+
@@ -45,7 +60,31 @@ Debug
- - +
+
+ + +
+
+
+

Player 1

+
+ W +
+
+ S +
+
+
+

Player 2

+
+ UP +
+
+ DOWN +
+
+
+
diff --git a/www/index.js b/www/index.js index 2e0d278..3711280 100644 --- a/www/index.js +++ b/www/index.js @@ -77,16 +77,15 @@ const tick = () => { let objects; if (!networkSession) { + renderActions(actions) field.tick(actions, update); objects = JSON.parse(field.objects()); } else if (isHost) { - // Would mean that input events would get lost if latency is higher than 100 ms. - const peerInputEvents = events.filter(e => e.topic === "input").filter(it => it.msg.player !== player.id) - const lastPeerInputEvents = peerInputEvents.length ? [peerInputEvents[peerInputEvents - 1]] : [] + const otherPlayerActions = getOtherPlayerInputActions(); const allActions = [ - ...actions, ...lastPeerInputEvents + ...actions, ...otherPlayerActions ]; - console.warn({allActions}) + renderActions(allActions); field.tick(allActions, update); objects = JSON.parse(field.objects()); sendEvents([...getInputEvents(), ...getMoveEvents(objects)]) @@ -99,6 +98,8 @@ const tick = () => { acc[objId].push(moveEvent); return acc; }, {}); + const otherPlayerActions = getOtherPlayerInputActions(); + renderActions([...actions, ...otherPlayerActions]); const latestMoveEvents = Object.entries(moveEventsByObj) .map(([_, moveEvents]) => moveEvents[moveEvents.length - 1]); objects = latestMoveEvents.map(({msg}) => msg); @@ -107,17 +108,22 @@ const tick = () => { render(objects); } +const getOtherPlayerInputActions = () => { + const peerInputEvents = events.filter(e => e.topic === "input").filter(it => it.msg.player !== player.id).map(it => it.msg) + const lastPeerInputEvent = peerInputEvents.length ? peerInputEvents[peerInputEvents.length - 1] : null + return lastPeerInputEvent ? lastPeerInputEvent.inputs : [] +} + const getMoveEvents = objects => { return objects.map(o => ({session_id: networkSession.hash, topic: 'move', msg: JSON.stringify({...o, session_id: networkSession.hash, ts: Date.now()})})); } const getInputEvents = () => { - const inputEvents = actions.map(({input}) => ({msg: JSON.stringify({inputs: [input], player: player.id, session_id: networkSession.hash, ts: Date.now()}), session_id: networkSession.hash, topic: 'input'})); - if (inputEvents.length) { - return inputEvents; + if (actions.length) { + const inputEvent = {msg: JSON.stringify({inputs: actions, player: player.id, session_id: networkSession.hash, ts: Date.now()}), session_id: networkSession.hash, topic: 'input'} + return [inputEvent] } - const noInputs = {inputs: [], obj_id: isHost ? 0 : 1, player: isHost ? 1 : 2 } - return [{msg: JSON.stringify({input: noInputs, player: player.id, session_id: networkSession.hash, ts: Date.now()}), session_id: networkSession.hash, topic: 'input'}]; + return [{msg: JSON.stringify({inputs: [], player: player.id, session_id: networkSession.hash, ts: Date.now()}), session_id: networkSession.hash, topic: 'input'}]; } const sendEvents = events => { @@ -132,6 +138,17 @@ const render = objects => { drawObjects(objects); } +const renderActions = actions => { + [...document.getElementsByClassName("game_input")].forEach(el => { + el.className = "game_input game_input--inactive" + }) + actions.forEach(action => { + const id = `game_inputs__player_${action.player}__${action.input.toLowerCase()}` + const el = document.getElementById(id); + el.className = "game_input game_input--active"; + }) +} + const reset = () => { framesLastSecond = []; lastFpsUpdate = 0;