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

session manager

Browse Source
This commit is contained in:
Thilo Behnke
2022-05-14 14:36:39 +02:00
parent ffc26e2f2f
commit 3ca0229667
6 changed files with 104 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
server/Cargo.toml
View File

@@ -10,7 +10,9 @@ kafka = { version = "0.8.0" }
hyper = {version = "0.14.18", features = ["full"]}
tokio = { version = "1", features = ["full"] }
tokio-stream = {version = "0.1" }
serde = { version = "1.0", features = ["derive"] }
serde_json = { version = "1.0.79" }
md5 = { version = "0.7.0" }
pong = { path = "../pong", version = "0.1.0" }
[dev-dependencies]

9
server/src/hash.rs Normal file
View File

@@ -0,0 +1,9 @@
use std::ptr::hash;
pub struct Hasher {}
impl Hasher {
pub fn hash(n: u32) -> String {
let digest = md5::compute(format!("{}", n));
format!("{:x}", digest)
}
}

25
server/src/http.rs
View File

@@ -1,38 +1,44 @@
use std::convert::Infallible;
use std::io::ErrorKind::NotFound;
use std::sync::Arc;
use hyper::{Body, body, Method, Request, Response, Server};
use hyper::{Body, body, Method, Request, Response, Server, StatusCode};
use hyper::server::conn::AddrStream;
use hyper::service::{make_service_fn, service_fn};
use kafka::producer::Producer;
use serde_json::json;
use tokio::sync::Mutex;
use pong::event::event::{Event, EventReader, EventWriter};
use crate::kafka::{KafkaEventReaderImpl, KafkaEventWriterImpl};
use crate::session::SessionManager;
use crate::utils::http_utils::get_query_params;
pub struct HttpServer {
addr: [u8; 4],
port: u16,
session_manager: Arc<Mutex<SessionManager>>,
event_writer: Arc<Mutex<EventWriter>>,
event_reader: Arc<Mutex<EventReader>>
}
impl HttpServer {
pub fn new(addr: [u8; 4], port: u16) -> HttpServer {
let session_manager = Arc::new(Mutex::new(SessionManager::new()));
let event_writer = Arc::new(Mutex::new(EventWriter::new(Box::new(KafkaEventWriterImpl::default()))));
let event_reader = Arc::new(Mutex::new(EventReader::new(Box::new(KafkaEventReaderImpl::default()))));
HttpServer {addr, port, event_writer, event_reader}
HttpServer {addr, port, session_manager, event_writer, event_reader}
}
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 mut event_writer = Arc::clone(&self.event_writer);
let mut event_reader = Arc::clone(&self.event_reader);
async move {
Ok::<_, Infallible>(service_fn(move |req: Request<Body>| {
let mut session_manager = Arc::clone(&session_manager);
let mut event_writer = Arc::clone(&event_writer);
let mut event_reader = Arc::clone(&event_reader);
async move {
return handle_request(&event_writer, &event_reader, req).await;
return handle_request(&session_manager, &event_writer, &event_reader, req).await;
}
}))
}
@@ -47,15 +53,26 @@ impl HttpServer {
}
}
async fn handle_request(event_writer: &Arc<Mutex<EventWriter>>, event_reader: &Arc<Mutex<EventReader>>, req: Request<Body>) -> Result<Response<Body>, Infallible> {
async fn handle_request(session_manager: &Arc<Mutex<SessionManager>>, event_writer: &Arc<Mutex<EventWriter>>, event_reader: &Arc<Mutex<EventReader>>, req: Request<Body>) -> 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, "/write") => handle_event_write(event_writer, req).await,
(&Method::GET, "/show") => handle_event_read(event_reader, req).await,
_ => Ok(Response::new("unknown".into()))
}
}
async fn handle_session_create(session_manager: &Arc<Mutex<SessionManager>>, req: Request<Body>) -> Result<Response<Body>, Infallible> {
let mut locked = session_manager.lock().await;
let session_create_res = locked.create_session().await;
if let Err(e) = session_create_res {
return Ok(Response::builder().status(StatusCode::INTERNAL_SERVER_ERROR).body(Body::from(e)).unwrap());
}
let serialized = json!(session_create_res.unwrap());
return Ok(Response::new(Body::from(serialized.to_string())))
}
async fn handle_event_write(event_writer: &Arc<Mutex<EventWriter>>, req: Request<Body>) -> Result<Response<Body>, Infallible> {
let mut locked = event_writer.lock().await;
let body = body::to_bytes(req.into_body()).await.unwrap();

35
server/src/kafka.rs
View File

@@ -1,6 +1,10 @@
use std::process::ExitStatus;
use std::time::Duration;
use kafka::client::KafkaClient;
use kafka::client::metadata::Topic;
use tokio::process::Command;
use kafka::consumer::{Consumer, FetchOffset, GroupOffsetStorage, MessageSet};
use kafka::producer::{DefaultPartitioner, Producer, Record, RequiredAcks};
use kafka::producer::{DefaultPartitioner, Producer, Record, RequiredAcks, Topics};
use pong::event::event::{Event, EventReaderImpl, EventWriter, EventWriterImpl};
pub struct KafkaEventWriterImpl {
@@ -87,3 +91,32 @@ impl KafkaEventReaderImpl {
events
}
}
#[derive(Debug)]
pub struct KafkaTopicManager {}
impl KafkaTopicManager {
pub async fn add_partition(&self) -> Result<u32, String> {
let topics = vec!["move", "status", "input"];
let mut client = KafkaClient::new(vec!["localhost:9092".to_owned()]);
client.load_metadata_all().unwrap();
let topic_metas: Vec<Topic> = client.topics().into_iter().filter(|t| topics.contains(&t.name())).collect();
if topic_metas.len() != 3 {
return Err(format!("Can't add_partition, unable to find matching topics: {:?}", topics));
}
let max_partition_count = topic_metas.into_iter().map(|t| t.partitions().len()).max().unwrap();
let next_partition = max_partition_count + 1;
println!("Will create next partition: {}", next_partition);
// TODO: What if creating one of the partitions fails?
for topic in topics {
let output = Command::new("/bin/bash").arg(format!("-c bin/kafka-topics.sh --bootstrap-server localhost:9092 --alter --topic {} --partitions {}", topic, next_partition)).output().await;
if let Err(e) = output {
return Err(format!("{}", e))
}
let output = output.unwrap();
if !output.status.success() {
return Err(format!("{:?}", std::str::from_utf8(&*output.stderr)))
}
}
Ok(next_partition as u32)
}
}

2
server/src/main.rs
View File

@@ -3,6 +3,8 @@ use crate::http::HttpServer;
pub mod http;
pub mod kafka;
pub mod utils;
mod hash;
mod session;
#[tokio::main]
pub async fn main() {

36
server/src/session.rs Normal file
View File

@@ -0,0 +1,36 @@
use crate::hash::Hasher;
use crate::kafka::KafkaTopicManager;
use serde::{Serialize};
#[derive(Debug)]
pub struct SessionManager {
sessions: Vec<Session>,
topic_manager: KafkaTopicManager
}
impl SessionManager {
pub fn new() -> SessionManager {
SessionManager {
sessions: vec![],
topic_manager: KafkaTopicManager {}
}
}
pub async fn create_session(&mut self) -> 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);
return Err(e);
}
let session_id = add_partition_res.unwrap();
let session_hash = Hasher::hash(session_id);
let session = Session {id: session_hash};
self.sessions.push(session.clone());
Ok(session)
}
}
#[derive(Debug, Clone, Serialize)]
pub struct Session {
id: String
}