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

working implementation for creating a session from the server

Browse Source
This commit is contained in:
Thilo Behnke
2022-05-28 20:07:36 +02:00
parent 750b5832d3
commit bd7c1832b8
6 changed files with 50 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
server/docker-compose.yml
View File

@@ -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

2
server/kafka/kafka-script-proxy/src/main.rs
View File

@@ -118,7 +118,7 @@ pub fn build_success_res(value: &str) -> Result<Response<Body>, Infallible> {
}
pub fn build_error_res(error: &str, status: StatusCode) -> Result<Response<Body>, 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);

6
server/src/http.rs
View File

@@ -20,10 +20,10 @@ pub struct HttpServer {
event_reader: Arc<Mutex<EventReader>>
}
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}
}

43
server/src/kafka.rs
View File

@@ -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<u32, String> {
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::<PartitionApiDTO>(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::<PartitionApiDTO>(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)
}
}

2
server/src/main.rs
View File

@@ -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");
}

3
server/src/session.rs
View File

@@ -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)
}