diff --git a/server/src/http.rs b/server/src/http.rs index c01e8ac..e83b7ff 100644 --- a/server/src/http.rs +++ b/server/src/http.rs @@ -1,6 +1,7 @@ use std::convert::Infallible; use std::fs::read; use std::io::ErrorKind::NotFound; +use std::net::SocketAddr; use std::sync::Arc; use hyper::{Body, body, Method, Request, Response, Server, StatusCode}; use hyper::server::conn::AddrStream; @@ -11,6 +12,7 @@ use serde::{Deserialize, Serialize}; use tokio::sync::Mutex; use pong::event::event::{Event, EventReader, EventWriter}; use crate::kafka::{KafkaEventReaderImpl, KafkaSessionEventWriterImpl}; +use crate::player::Player; use crate::session::{CachingSessionManager, SessionManager}; use crate::utils::http_utils::{get_query_params, read_json_body}; @@ -28,11 +30,12 @@ impl HttpServer { 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(); async move { Ok::<_, Infallible>(service_fn(move |req: Request| { let mut session_manager = Arc::clone(&session_manager); async move { - return handle_request(&session_manager, req).await; + return handle_request(&session_manager, req, addr).await; } })) } @@ -47,23 +50,20 @@ impl HttpServer { } } -async fn handle_request(session_manager: &Arc>, req: Request) -> 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::POST, "/create_session") => handle_session_create(session_manager, req).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())) } } -// TODO: Both for write and read session: -// - use session id from req body -// - pass event write / read to session manager that holds references to session specific readers/writers - -async fn handle_session_create(session_manager: &Arc>, req: Request) -> Result, Infallible> { +async fn handle_session_create(session_manager: &Arc>, req: Request, addr: SocketAddr) -> Result, Infallible> { let mut locked = session_manager.lock().await; - let session_create_res = locked.create_session().await; + let session_create_res = locked.create_session(Player {id: addr.to_string()}).await; if let Err(e) = session_create_res { return Ok(Response::builder().status(StatusCode::INTERNAL_SERVER_ERROR).body(Body::from(e)).unwrap()); } @@ -71,6 +71,17 @@ async fn handle_session_create(session_manager: &Arc>, mut req: Request, addr: SocketAddr) -> Result, Infallible> { + let mut locked = session_manager.lock().await; + let body = read_json_body::(&mut req).await; + let session_join_res = locked.join_session(body.session_id, Player {id: addr.to_string()}).await; + if let Err(e) = session_join_res { + return Ok(Response::builder().status(StatusCode::INTERNAL_SERVER_ERROR).body(Body::from(e)).unwrap()); + } + let serialized = json!(session_join_res.unwrap()); + return Ok(Response::new(Body::from(serialized.to_string()))) +} + 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; @@ -147,3 +158,8 @@ struct SessionEventWriteDTO { struct SessionReadDTO { session_id: String } + +#[derive(Debug, Serialize, Deserialize)] +struct SessionJoinDto { + session_id: String +} diff --git a/server/src/main.rs b/server/src/main.rs index 903e9c0..e4fb33c 100644 --- a/server/src/main.rs +++ b/server/src/main.rs @@ -7,6 +7,7 @@ pub mod kafka; pub mod utils; mod hash; mod session; +mod player; #[tokio::main] pub async fn main() { diff --git a/server/src/player.rs b/server/src/player.rs new file mode 100644 index 0000000..2fce89f --- /dev/null +++ b/server/src/player.rs @@ -0,0 +1,7 @@ +use hyper::{Body, Request}; +use serde::{Deserialize, Serialize}; + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct Player { + pub id: String +} diff --git a/server/src/session.rs b/server/src/session.rs index 72f69c6..d0d31bd 100644 --- a/server/src/session.rs +++ b/server/src/session.rs @@ -5,9 +5,11 @@ 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, @@ -27,7 +29,7 @@ impl SessionManager { } } - pub async fn create_session(&mut self) -> Result { + pub async fn create_session(&mut self, player: Player) -> Result { let add_partition_res = self.topic_manager.add_partition().await; if let Err(e) = add_partition_res { println!("Failed to create partition: {}", e); @@ -35,9 +37,42 @@ impl SessionManager { } let session_id = add_partition_res.unwrap(); let session_hash = Hasher::hash(session_id); - let session = Session {id: session_id, hash: session_hash}; + let session = Session::new (session_id, session_hash, player.clone()); println!("Successfully created session: {:?}", session); - let json_event = serde_json::to_string(&SessionEvent::created(session.clone())).unwrap(); + 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 { + let updated_session = { + let session = self.sessions.iter_mut().find(|s| s.hash == session_id); + if let None = session { + let error = format!("Can't join session that does not exist: {}", session_id); + return Err(error); + } + let mut session = session.unwrap(); + if session.state != SessionState::PENDING { + let error = format!("Can't join session that is not PENDING: {}", session_id); + return Err(error); + } + if session.players.len() > 1 { + let error = format!("Can't join session with more than 1 player: {}", session_id); + return Err(error); + } + session.players.push(player.clone()); + session.state = SessionState::RUNNING; + session.clone() + }; + { + self.write_to_producer(session_joined(updated_session.clone(), player.clone())); + }; + println!("sessions = {:?}", self.sessions); + Ok(updated_session.clone()) + } + + 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}); if let Err(e) = session_event_write { let message = format!("Failed to write session create event to kafka: {:?}", e); @@ -45,8 +80,7 @@ impl SessionManager { return Err(message.to_owned()); } println!("Successfully produced session event."); - self.sessions.push(session.clone()); - Ok(session) + return Ok(()); } pub fn get_session_reader(&self, session_id: &str) -> Result { @@ -89,8 +123,12 @@ impl CachingSessionManager { } } - pub async fn create_session(&mut self) -> Result { - self.inner.create_session().await + pub async fn create_session(&mut self, player: Player) -> Result { + self.inner.create_session(player).await + } + + 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> { @@ -127,7 +165,39 @@ impl CachingSessionManager { #[derive(Debug, Clone, Serialize, Deserialize)] pub struct Session { pub id: u16, - pub hash: String + pub hash: String, + pub state: SessionState, + players: Vec +} + +impl Session { + pub fn new(id: u16, hash: String, player: Player) -> Session { + Session { + players: vec!(player), + id, + hash, + state: SessionState::PENDING + } + } + + pub fn can_be_joined(&self) -> bool { + self.players.len() == 1 + } + + pub fn join(&mut self, player: Player) -> bool { + if !self.can_be_joined() { + return false + } + self.players.push(player); + return true; + } +} + +#[derive(Clone, Debug, Serialize, Deserialize, Eq, PartialEq)] +pub enum SessionState { + PENDING, // 1 player is missing + RUNNING, // game is playing + CLOSED // game is over } pub struct SessionWriter { @@ -154,18 +224,48 @@ impl SessionReader { } #[derive(Deserialize, Serialize)] -struct SessionEvent { +struct SessionCreatedEvent { event_type: SessionEventType, - session: Session + session: Session, + player: Player } -impl SessionEvent { - pub fn created(session: Session) -> SessionEvent { - SessionEvent {event_type: SessionEventType::CREATED, session} +impl SessionCreatedEvent { + pub fn new(session: Session, player: Player) -> SessionCreatedEvent { + SessionCreatedEvent { + event_type: SessionEventType::CREATED, + session, + player + } } } #[derive(Deserialize, Serialize)] -enum SessionEventType { - CREATED +struct SessionJoinedEvent { + event_type: SessionEventType, + session: Session, + player: Player +} + +impl SessionJoinedEvent { + pub fn new(session: Session, player: Player) -> SessionJoinedEvent { + SessionJoinedEvent { + event_type: SessionEventType::JOINED, + session, + player + } + } +} + +fn session_created(session: Session, player: Player) -> SessionCreatedEvent { + SessionCreatedEvent::new(session, player) +} + +fn session_joined(session: Session, player: Player) -> SessionJoinedEvent { + SessionJoinedEvent::new(session, player) +} + +#[derive(Deserialize, Serialize)] +enum SessionEventType { + CREATED, JOINED }