diff --git a/server/Cargo.toml b/server/Cargo.toml index 27f2a4b..d4c426a 100644 --- a/server/Cargo.toml +++ b/server/Cargo.toml @@ -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] diff --git a/server/src/hash.rs b/server/src/hash.rs new file mode 100644 index 0000000..1c7be62 --- /dev/null +++ b/server/src/hash.rs @@ -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) + } +} diff --git a/server/src/http.rs b/server/src/http.rs index 9b3032d..bc4a1e1 100644 --- a/server/src/http.rs +++ b/server/src/http.rs @@ -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>, event_writer: Arc>, event_reader: Arc> } 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> { 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| { + 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>, event_reader: &Arc>, req: Request) -> Result, Infallible> { +async fn handle_request(session_manager: &Arc>, event_writer: &Arc>, event_reader: &Arc>, req: Request) -> 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, "/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>, req: Request) -> Result, 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>, req: Request) -> Result, Infallible> { let mut locked = event_writer.lock().await; let body = body::to_bytes(req.into_body()).await.unwrap(); diff --git a/server/src/kafka.rs b/server/src/kafka.rs index 5e4b2dd..1396e58 100644 --- a/server/src/kafka.rs +++ b/server/src/kafka.rs @@ -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 { + 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 = 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) + } +} diff --git a/server/src/main.rs b/server/src/main.rs index 720d55f..748c02e 100644 --- a/server/src/main.rs +++ b/server/src/main.rs @@ -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() { diff --git a/server/src/session.rs b/server/src/session.rs new file mode 100644 index 0000000..9bae57c --- /dev/null +++ b/server/src/session.rs @@ -0,0 +1,36 @@ +use crate::hash::Hasher; +use crate::kafka::KafkaTopicManager; +use serde::{Serialize}; + +#[derive(Debug)] +pub struct SessionManager { + sessions: Vec, + topic_manager: KafkaTopicManager +} + +impl SessionManager { + pub fn new() -> SessionManager { + SessionManager { + sessions: vec![], + topic_manager: KafkaTopicManager {} + } + } + + pub async fn create_session(&mut self) -> Result { + 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 +}