From bd7c1832b85c45de8be03e4ba2d2e2940ca408a2 Mon Sep 17 00:00:00 2001 From: Thilo Behnke Date: Sat, 28 May 2022 20:07:36 +0200 Subject: [PATCH] working implementation for creating a session from the server --- server/docker-compose.yml | 19 ++++----- server/kafka/kafka-script-proxy/src/main.rs | 2 +- server/src/http.rs | 6 +-- server/src/kafka.rs | 43 ++++++++++++++++----- server/src/main.rs | 2 +- server/src/session.rs | 3 +- 6 files changed, 50 insertions(+), 25 deletions(-) diff --git a/server/docker-compose.yml b/server/docker-compose.yml index 6408902..3217f09 100644 --- a/server/docker-compose.yml +++ b/server/docker-compose.yml @@ -13,6 +13,7 @@ services: context: kafka ports: - '9092:9092' + - '9093:9093' - '7243:7243' environment: - KAFKA_BROKER_ID=1 @@ -24,13 +25,13 @@ services: - ALLOW_PLAINTEXT_LISTENER=yes depends_on: - zookeeper - kaka-rest-proxy: - container_name: pong_server_kafka_rest_proxy - image: 'confluentinc/cp-kafka-rest' - ports: - - '8082:8082' - environment: - - KAFKA_REST_BOOTSTRAP_SERVERS=kafka:9092 - depends_on: - - kafka +# kaka-rest-proxy: +# container_name: pong_server_kafka_rest_proxy +# image: 'confluentinc/cp-kafka-rest' +# ports: +# - '8082:8082' +# environment: +# - KAFKA_REST_BOOTSTRAP_SERVERS=kafka:9092 +# depends_on: +# - kafka diff --git a/server/kafka/kafka-script-proxy/src/main.rs b/server/kafka/kafka-script-proxy/src/main.rs index b248d5e..ff5c865 100644 --- a/server/kafka/kafka-script-proxy/src/main.rs +++ b/server/kafka/kafka-script-proxy/src/main.rs @@ -118,7 +118,7 @@ pub fn build_success_res(value: &str) -> Result, Infallible> { } pub fn build_error_res(error: &str, status: StatusCode) -> Result, Infallible> { - let json = format!("{{\"error\": {}}}", error); + let json = format!("{{\"error\": \"{}\"}}", error); let mut res = Response::new(Body::from(json)); *res.status_mut() = status; return Ok(res); diff --git a/server/src/http.rs b/server/src/http.rs index bc4a1e1..3d0796a 100644 --- a/server/src/http.rs +++ b/server/src/http.rs @@ -20,10 +20,10 @@ pub struct HttpServer { event_reader: Arc> } impl HttpServer { - pub fn new(addr: [u8; 4], port: u16) -> HttpServer { + pub fn new(addr: [u8; 4], port: u16, kafka_host: &str) -> 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())))); + let event_writer = Arc::new(Mutex::new(EventWriter::new(Box::new(KafkaEventWriterImpl::from(kafka_host))))); + let event_reader = Arc::new(Mutex::new(EventReader::new(Box::new(KafkaEventReaderImpl::from(kafka_host))))); HttpServer {addr, port, session_manager, event_writer, event_reader} } diff --git a/server/src/kafka.rs b/server/src/kafka.rs index 7c5e9c9..f2ebb95 100644 --- a/server/src/kafka.rs +++ b/server/src/kafka.rs @@ -1,7 +1,7 @@ use std::process::ExitStatus; use std::str::FromStr; use std::time::Duration; -use hyper::{Client, Uri}; +use hyper::{Body, Client, Method, Request, Uri}; use serde::{Deserialize}; use kafka::client::KafkaClient; use kafka::client::metadata::Topic; @@ -15,10 +15,15 @@ pub struct KafkaEventWriterImpl { } impl KafkaEventWriterImpl { pub fn default() -> KafkaEventWriterImpl { - KafkaEventWriterImpl::new("localhost:9092") + KafkaEventWriterImpl::new("localhost:9093") + } + + pub fn from(host: &str) -> KafkaEventWriterImpl { + KafkaEventWriterImpl::new(host) } pub fn new(host: &str) -> KafkaEventWriterImpl { + println!("Connecting producer to kafka host: {}", host); let mut producer = Producer::from_hosts(vec![host.to_owned()]) .with_ack_timeout(Duration::from_secs(1)) .with_required_acks(RequiredAcks::One) @@ -44,11 +49,16 @@ pub struct KafkaEventReaderImpl { } impl KafkaEventReaderImpl { pub fn default() -> KafkaEventReaderImpl { - KafkaEventReaderImpl::new("localhost:9092") + KafkaEventReaderImpl::new("localhost:9093") + } + + pub fn from(host: &str) -> KafkaEventReaderImpl { + KafkaEventReaderImpl::new(host) } pub fn new(host: &str) -> KafkaEventReaderImpl { - let mut consumer = Consumer::from_hosts(vec![host.to_owned()]) + println!("Connecting consumer to kafka host: {}", host); + 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()) @@ -102,17 +112,23 @@ pub struct KafkaTopicManager { impl KafkaTopicManager { pub fn default() -> KafkaTopicManager { - KafkaTopicManager {partition_management_endpoint: "kafka: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()} } pub async fn add_partition(&self) -> Result { let mut client = Client::new(); - let res = client.get(Uri::from_str(&self.partition_management_endpoint).unwrap()).await; + 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); println!("{}", error); return Err(error); } + let status = res.as_ref().unwrap().status(); let bytes = hyper::body::to_bytes(res.unwrap()).await; if let Err(e) = bytes { let error = format!("Failed to read bytes from response: {:?}", e); @@ -126,13 +142,20 @@ impl KafkaTopicManager { println!("{}", error); return Err(error); } - let json = serde_json::from_str::(res_str.unwrap()); - if let Err(e) = json { - let error = format!("Failed to convert string to json: {:?}", e); + if status != 200 { + let error = format!("Failed to add partition: {}", res_str.unwrap()); println!("{}", error); return Err(error); } - Ok(json.unwrap().data) + 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); + println!("{}", error); + return Err(error); + } + let updated_partition_count = json.unwrap().data; + println!("Successfully created partition: {}", updated_partition_count); + Ok(updated_partition_count) } } diff --git a/server/src/main.rs b/server/src/main.rs index 491e497..903e9c0 100644 --- a/server/src/main.rs +++ b/server/src/main.rs @@ -10,5 +10,5 @@ mod session; #[tokio::main] pub async fn main() { - HttpServer::new([127, 0, 0, 1], 4000).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/session.rs b/server/src/session.rs index 38e8b1f..73b838f 100644 --- a/server/src/session.rs +++ b/server/src/session.rs @@ -12,7 +12,7 @@ impl SessionManager { pub fn new() -> SessionManager { SessionManager { sessions: vec![], - topic_manager: KafkaTopicManager::default() + topic_manager: KafkaTopicManager::from("localhost:7243") } } @@ -25,6 +25,7 @@ impl SessionManager { let session_id = add_partition_res.unwrap(); let session_hash = Hasher::hash(session_id); let session = Session {id: session_hash}; + println!("Successfully created session: {:?}", session); self.sessions.push(session.clone()); Ok(session) }