github/wasm-pong
1
0
Fork 0
mirror of https://github.com/thilo-behnke/wasm-pong.git synced 2026-07-12 19:19:16 +00:00
Code Issues Packages Projects Releases Wiki Activity

join session

Browse Source
This commit is contained in:
Thilo Behnke
2022-05-29 15:49:54 +02:00
parent 35a4721c75
commit 9e7d25fbd4
4 changed files with 148 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
server/src/http.rs
View File

@@ -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<dyn std::error::Error + Send + Sync>> {
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<Body>| {
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<Mutex<CachingSessionManager>>, req: Request<Body>) -> Result<Response<Body>, Infallible> {
async fn handle_request(session_manager: &Arc<Mutex<CachingSessionManager>>, req: Request<Body>, addr: SocketAddr) -> Result<Response<Body>, 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<Mutex<CachingSessionManager>>, req: Request<Body>) -> Result<Response<Body>, Infallible> {
async fn handle_session_create(session_manager: &Arc<Mutex<CachingSessionManager>>, req: Request<Body>, addr: SocketAddr) -> Result<Response<Body>, 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<Mutex<CachingSessionManager
return Ok(Response::new(Body::from(serialized.to_string())))
}
async fn handle_session_join(session_manager: &Arc<Mutex<CachingSessionManager>>, mut req: Request<Body>, addr: SocketAddr) -> Result<Response<Body>, Infallible> {
let mut locked = session_manager.lock().await;
let body = read_json_body::<SessionJoinDto>(&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<Mutex<CachingSessionManager>>, mut req: Request<Body>) -> Result<Response<Body>, Infallible> {
let mut locked = session_manager.lock().await;
let event = read_json_body::<SessionEventWriteDTO>(&mut req).await;
@@ -147,3 +158,8 @@ struct SessionEventWriteDTO {
struct SessionReadDTO {
session_id: String
}
#[derive(Debug, Serialize, Deserialize)]
struct SessionJoinDto {
session_id: String
}

1
server/src/main.rs
View File

@@ -7,6 +7,7 @@ pub mod kafka;
pub mod utils;
mod hash;
mod session;
mod player;
#[tokio::main]
pub async fn main() {

7
server/src/player.rs Normal file
View File

@@ -0,0 +1,7 @@
use hyper::{Body, Request};
use serde::{Deserialize, Serialize};
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Player {
pub id: String
}

130
server/src/session.rs
View File

@@ -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<Session, String> {
pub async fn create_session(&mut self, player: Player) -> Result<Session, String> {
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<Session, String> {
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<T>(&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<SessionReader, String> {
@@ -89,8 +123,12 @@ impl CachingSessionManager {
}
}
pub async fn create_session(&mut self) -> Result<Session, String> {
self.inner.create_session().await
pub async fn create_session(&mut self, player: Player) -> Result<Session, String> {
self.inner.create_session(player).await
}
pub async fn join_session(&mut self, session_id: String, player: Player) -> Result<Session, String> {
self.inner.join_session(session_id, player).await
}
pub fn get_session_reader(&mut self, session_id: &str) -> Result<Arc<Mutex<SessionReader>>, 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<Player>
}
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
}