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

working input transmission

Browse Source
This commit is contained in:
Thilo Behnke
2022-06-11 15:20:32 +02:00
parent 26316e6f47
commit b77ce89e80
22 changed files with 671 additions and 368 deletions
Download Patch File Download Diff File Expand all files Collapse all files

263
server/src/http.rs
View File

@@ -1,3 +1,18 @@
use crate::kafka::{KafkaEventReaderImpl, KafkaSessionEventWriterImpl};
use crate::player::Player;
use crate::session::{Session, SessionManager};
use crate::utils::http_utils::{get_query_params, read_json_body};
use crate::utils::time_utils::now;
use futures::{sink::SinkExt, stream::StreamExt};
use hyper::server::conn::AddrStream;
use hyper::service::{make_service_fn, service_fn};
use hyper::{body, Body, Method, Request, Response, Server, StatusCode};
use hyper_tungstenite::tungstenite::{Error, Message};
use hyper_tungstenite::{tungstenite, HyperWebsocket};
use kafka::producer::Producer;
use pong::event::event::{Event, EventReader, EventWriter};
use serde::{Deserialize, Serialize};
use serde_json::json;
use std::convert::Infallible;
use std::fs::read;
use std::io::ErrorKind::NotFound;
@@ -5,38 +20,27 @@ use std::net::SocketAddr;
use std::str::FromStr;
use std::sync::Arc;
use std::time::{Duration, SystemTime};
use hyper::{Body, body, Method, Request, Response, Server, StatusCode};
use hyper::server::conn::AddrStream;
use hyper::service::{make_service_fn, service_fn};
use hyper_tungstenite::{HyperWebsocket, tungstenite};
use hyper_tungstenite::tungstenite::{Error, Message};
use kafka::producer::Producer;
use serde_json::json;
use serde::{Deserialize, Serialize};
use tokio::sync::Mutex;
use pong::event::event::{Event, EventReader, EventWriter};
use futures::{sink::SinkExt, stream::StreamExt};
use tokio::io::Sink;
use tokio::sync::Mutex;
use tokio::task;
use tokio::time::sleep;
use crate::kafka::{KafkaEventReaderImpl, KafkaSessionEventWriterImpl};
use crate::player::Player;
use crate::session::{Session, SessionManager};
use crate::utils::http_utils::{get_query_params, read_json_body};
use crate::utils::time_utils::now;
pub struct HttpServer {
addr: [u8; 4],
port: u16,
session_manager: Arc<Mutex<SessionManager>>
session_manager: Arc<Mutex<SessionManager>>,
}
impl HttpServer {
pub fn new(addr: [u8; 4], port: u16, kafka_host: &str) -> HttpServer {
let session_manager = Arc::new(Mutex::new(SessionManager::new(kafka_host)));
HttpServer {addr, port, session_manager}
HttpServer {
addr,
port,
session_manager,
}
}
pub async fn run(self) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
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();
@@ -45,23 +49,34 @@ impl HttpServer {
let mut session_manager = Arc::clone(&session_manager);
async move {
if hyper_tungstenite::is_upgrade_request(&req) {
println!("Received request to upgrade to websocket connection: {:?}", req);
println!(
"Received request to upgrade to websocket connection: {:?}",
req
);
let params = get_query_params(&req);
println!("Ws request params: {:?}", params);
if !params.contains_key("session_id") {
eprintln!("Missing session id request param for websocket connection, don't upgrade connection to ws.");
return build_error_res("Missing request param: session_id", StatusCode::BAD_REQUEST);
return build_error_res(
"Missing request param: session_id",
StatusCode::BAD_REQUEST,
);
}
if !params.contains_key("connection_type") {
eprintln!("Missing connection type request param for websocket connection, don't upgrade connection to ws.");
let res = build_error_res("Missing request param: connection_type", StatusCode::BAD_REQUEST);
let res = build_error_res(
"Missing request param: connection_type",
StatusCode::BAD_REQUEST,
);
return res;
}
let session_id = params.get("session_id").unwrap();
let connection_type_raw = params.get("connection_type").unwrap();
let connection_type = WebSocketConnectionType::from_str(connection_type_raw);
let connection_type =
WebSocketConnectionType::from_str(connection_type_raw);
if let Err(_) = connection_type {
let error = format!("Invalid connection type: {}", connection_type_raw);
let error =
format!("Invalid connection type: {}", connection_type_raw);
eprintln!("{}", error);
return build_error_res(error.as_str(), StatusCode::BAD_REQUEST);
}
@@ -72,20 +87,27 @@ impl HttpServer {
return build_error_res(error.as_str(), StatusCode::NOT_FOUND);
}
let session = session.unwrap();
let websocket_session = WebSocketSession {session: session.clone(), connection_type: connection_type.unwrap()};
let websocket_session = WebSocketSession {
session: session.clone(),
connection_type: connection_type.unwrap(),
};
println!("Websocket upgrade request is valid, will now upgrade to websocket: {:?}", req);
let (response, websocket) = hyper_tungstenite::upgrade(req, None).unwrap();
let (response, websocket) =
hyper_tungstenite::upgrade(req, None).unwrap();
// Spawn a task to handle the websocket connection.
tokio::spawn(async move {
if let Err(e) = serve_websocket(websocket_session, websocket, session_manager).await {
if let Err(e) =
serve_websocket(websocket_session, websocket, session_manager)
.await
{
eprintln!("Error in websocket connection: {:?}", e);
}
});
// Return the response so the spawned future can continue.
return Ok(response)
return Ok(response);
}
return handle_request(&session_manager, req, addr).await;
@@ -104,22 +126,34 @@ impl HttpServer {
}
/// Handle a websocket connection.
async fn serve_websocket(websocket_session: WebSocketSession, websocket: HyperWebsocket, session_manager: Arc<Mutex<SessionManager>>) -> Result<(), Error> {
async fn serve_websocket(
websocket_session: WebSocketSession,
websocket: HyperWebsocket,
session_manager: Arc<Mutex<SessionManager>>,
) -> Result<(), Error> {
let mut websocket = websocket.await?;
let (mut websocket_writer, mut websocket_reader) = websocket.split();
let session_manager = session_manager.lock().await;
let session = session_manager.get_session(&websocket_session.session.hash);
let event_handler_pair = session_manager.split(&websocket_session.session.hash, websocket_session.connection_type.get_topics());
let event_handler_pair = session_manager.split(
&websocket_session.session.hash,
websocket_session.connection_type.get_topics(),
);
if let Err(_) = event_handler_pair {
eprintln!("Failed to create event reader/writer pair session: {:?}", websocket_session);
return Err(Error::ConnectionClosed) // TODO: Use proper error for this case to close the connection
eprintln!(
"Failed to create event reader/writer pair session: {:?}",
websocket_session
);
return Err(Error::ConnectionClosed); // TODO: Use proper error for this case to close the connection
}
let (mut event_reader, mut event_writer) = event_handler_pair.unwrap();
let websocket_session_read_copy = websocket_session.clone();
tokio::spawn(async move {
println!("Ready to read messages from ws connection: {:?}", websocket_session_read_copy);
println!(
"Ready to read messages from ws connection: {:?}",
websocket_session_read_copy
);
while let Some(message) = websocket_reader.next().await {
match message.unwrap() {
Message::Text(msg) => {
@@ -144,22 +178,31 @@ async fn serve_websocket(websocket_session: WebSocketSession, websocket: HyperWe
}
}
if any_error {
eprintln!("Failed to write at least one message for session {}", event_wrapper.session_id);
eprintln!(
"Failed to write at least one message for session {}",
event_wrapper.session_id
);
} else {
println!("Successfully wrote {} messages to kafka for session {:?}", event_count, websocket_session_read_copy)
println!(
"Successfully wrote {} messages to kafka for session {:?}",
event_count, websocket_session_read_copy
)
}
},
}
Message::Close(msg) => {
// No need to send a reply: tungstenite takes care of this for you.
if let Some(msg) = &msg {
println!("Received close message with code {} and message: {}", msg.code, msg.reason);
println!(
"Received close message with code {} and message: {}",
msg.code, msg.reason
);
} else {
println!("Received close message");
}
let session_closed_event = SessionClosedDto {
session: websocket_session_read_copy.session.clone(),
reason: "ws closed".to_owned()
reason: "ws closed".to_owned(),
};
let msg = json!(session_closed_event).to_string();
let session_event_write_res = event_writer.write_to_session("session", &msg);
@@ -167,7 +210,7 @@ async fn serve_websocket(websocket_session: WebSocketSession, websocket: HyperWe
eprintln!("Failed to write session closed event: {0}", e)
}
break;
},
}
_ => {}
}
}
@@ -175,38 +218,41 @@ async fn serve_websocket(websocket_session: WebSocketSession, websocket: HyperWe
});
let websocket_session_write_copy = websocket_session.clone();
tokio::spawn(async move {
println!("Ready to read messages from kafka: {:?}", websocket_session_write_copy);
println!(
"Ready to read messages from kafka: {:?}",
websocket_session_write_copy
);
loop {
println!("Reading messages from kafka.");
let messages = event_reader.read_from_session();
if let Err(_) = messages {
eprintln!("Failed to read messages from kafka for session: {:?}", websocket_session_write_copy);
eprintln!(
"Failed to read messages from kafka for session: {:?}",
websocket_session_write_copy
);
continue;
}
// println!("Read messages for websocket_session {:?} from consumer: {:?}", websocket_session_write_copy, messages);
let messages = messages.unwrap();
if messages.len() == 0 {
println!("No new messages from kafka.");
continue;
}
println!("{} new messages from kafka.", messages.len());
let json = serde_json::to_string(&messages).unwrap();
let message = Message::from(json);
println!("Sending kafka messages through websocket.");
let send_res = websocket_writer.send(message).await;
if let Err(e) = send_res {
eprintln!("Failed to send message to websocket for session {:?}: {:?}", websocket_session_write_copy, e);
match e {
tungstenite::error::Error::ConnectionClosed | tungstenite::error::Error::AlreadyClosed => {
println!("Websocket Connection for session {:?} is closed. Exiting kafka consumer.", websocket_session_write_copy);
break;
},
_ => {}
} else {
println!("{} new messages from kafka.", messages.len());
let json = serde_json::to_string(&messages).unwrap();
let message = Message::from(json);
println!("Sending kafka messages through websocket.");
let send_res = websocket_writer.send(message).await;
if let Err(e) = send_res {
eprintln!(
"Failed to send message to websocket for session {:?}: {:?}",
websocket_session_write_copy, e
);
break;
}
}
// Avoid starvation of read thread (?)
// TODO: How to avoid this? This is very bad for performance.
sleep(Duration::from_millis(100)).await;
sleep(Duration::from_millis(1)).await;
}
});
Ok(())
@@ -214,19 +260,28 @@ async fn serve_websocket(websocket_session: WebSocketSession, websocket: HyperWe
// TODO: How to handle event writes/reads? This must be a websocket, but how to implement in hyper (if possible)?
// https://github.com/de-vri-es/hyper-tungstenite-rs
async fn handle_request(session_manager: &Arc<Mutex<SessionManager>>, req: Request<Body>, addr: SocketAddr) -> Result<Response<Body>, Infallible> {
async fn handle_request(
session_manager: &Arc<Mutex<SessionManager>>,
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::GET, "/session") => handle_get_session(session_manager, req).await,
(&Method::POST, "/create_session") => handle_session_create(session_manager, req, addr).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()))
_ => Ok(Response::new("unknown".into())),
}
}
async fn handle_get_session(session_manager: &Arc<Mutex<SessionManager>>, req: Request<Body>) -> Result<Response<Body>, Infallible> {
async fn handle_get_session(
session_manager: &Arc<Mutex<SessionManager>>,
req: Request<Body>,
) -> Result<Response<Body>, Infallible> {
let mut locked = session_manager.lock().await;
let query_params = get_query_params(&req);
let session_id = query_params.get("session_id");
@@ -241,37 +296,61 @@ async fn handle_get_session(session_manager: &Arc<Mutex<SessionManager>>, req: R
return build_success_res(&serde_json::to_string(&session.unwrap()).unwrap());
}
async fn handle_session_create(session_manager: &Arc<Mutex<SessionManager>>, req: Request<Body>, addr: SocketAddr) -> Result<Response<Body>, Infallible> {
async fn handle_session_create(
session_manager: &Arc<Mutex<SessionManager>>,
req: Request<Body>,
addr: SocketAddr,
) -> Result<Response<Body>, Infallible> {
println!("Called to create new session: {:?}", req);
let mut locked = session_manager.lock().await;
let player = Player {id: addr.to_string()};
let player = Player {
id: addr.to_string(),
};
let session_create_res = locked.create_session(player.clone()).await;
if let Err(e) = session_create_res {
return Ok(Response::builder().status(StatusCode::INTERNAL_SERVER_ERROR).body(Body::from(e)).unwrap());
return Ok(Response::builder()
.status(StatusCode::INTERNAL_SERVER_ERROR)
.body(Body::from(e))
.unwrap());
}
let session_created = SessionCreatedDto {session: session_create_res.unwrap(), player};
let session_created = SessionCreatedDto {
session: session_create_res.unwrap(),
player,
};
let serialized = json!(session_created);
return build_success_res(&serialized.to_string());
}
async fn handle_session_join(session_manager: &Arc<Mutex<SessionManager>>, mut req: Request<Body>, addr: SocketAddr) -> Result<Response<Body>, Infallible> {
async fn handle_session_join(
session_manager: &Arc<Mutex<SessionManager>>,
mut req: Request<Body>,
addr: SocketAddr,
) -> Result<Response<Body>, Infallible> {
println!("Received request to join session: {:?}", req);
let mut locked = session_manager.lock().await;
let body = read_json_body::<SessionJoinDto>(&mut req).await;
let player = Player {id: addr.to_string()};
let player = Player {
id: addr.to_string(),
};
let session_join_res = locked.join_session(body.session_id, player.clone()).await;
if let Err(e) = session_join_res {
eprintln!("Failed to join session: {:?}", e);
return Ok(Response::builder().status(StatusCode::INTERNAL_SERVER_ERROR).body(Body::from(e)).unwrap());
return Ok(Response::builder()
.status(StatusCode::INTERNAL_SERVER_ERROR)
.body(Body::from(e))
.unwrap());
}
let session = session_join_res.unwrap();
println!("Successfully joined session: {:?}", session);
let session_joined = SessionJoinedDto {session, player};
let session_joined = SessionJoinedDto { session, player };
let serialized = json!(session_joined);
return build_success_res(&serialized.to_string());
}
async fn handle_event_write(session_manager: &Arc<Mutex<SessionManager>>, mut req: Request<Body>) -> Result<Response<Body>, Infallible> {
async fn handle_event_write(
session_manager: &Arc<Mutex<SessionManager>>,
mut req: Request<Body>,
) -> Result<Response<Body>, Infallible> {
let mut locked = session_manager.lock().await;
let event = read_json_body::<SessionEventWriteDTO>(&mut req).await;
let writer = locked.get_session_writer(&event.session_id);
@@ -296,10 +375,16 @@ async fn handle_event_write(session_manager: &Arc<Mutex<SessionManager>>, mut re
build_success_res(&serde_json::to_string(&event).unwrap())
}
async fn handle_event_read(session_manager: &Arc<Mutex<SessionManager>>, mut req: Request<Body>) -> Result<Response<Body>, Infallible> {
async fn handle_event_read(
session_manager: &Arc<Mutex<SessionManager>>,
mut req: Request<Body>,
) -> Result<Response<Body>, Infallible> {
let mut locked = session_manager.lock().await;
let read_payload = read_json_body::<SessionReadDTO>(&mut req).await;
let reader = locked.get_session_reader(&read_payload.session_id, &["move", "status", "input", "session"]);
let reader = locked.get_session_reader(
&read_payload.session_id,
&["move", "status", "input", "session"],
);
if let Err(e) = reader {
let err = format!("Failed to read events: {}", e);
println!("{}", err);
@@ -308,7 +393,10 @@ async fn handle_event_read(session_manager: &Arc<Mutex<SessionManager>>, mut req
return Ok(res);
}
let mut reader = reader.unwrap();
println!("Reading session events from kafka for session: {}", read_payload.session_id);
println!(
"Reading session events from kafka for session: {}",
read_payload.session_id
);
let events = reader.read_from_session();
if let Err(e) = events {
let err = format!("Failed to read events: {}", e);
@@ -327,7 +415,10 @@ pub fn build_success_res(value: &str) -> Result<Response<Body>, Infallible> {
let mut res = Response::new(Body::from(json));
let headers = res.headers_mut();
headers.insert("Content-Type", "application/json".parse().unwrap());
headers.insert("Access-Control-Allow-Origin", "http://localhost:8080".parse().unwrap());
headers.insert(
"Access-Control-Allow-Origin",
"http://localhost:8080".parse().unwrap(),
);
Ok(res)
}
@@ -348,24 +439,24 @@ async fn shutdown_signal() {
#[derive(Debug, Deserialize, Serialize)]
struct SessionEventListDTO {
session_id: String,
events: Vec<SessionEventWriteDTO>
events: Vec<SessionEventWriteDTO>,
}
#[derive(Debug, Deserialize, Serialize)]
struct SessionEventWriteDTO {
session_id: String,
topic: String,
msg: String
msg: String,
}
#[derive(Debug, Serialize, Deserialize)]
struct SessionReadDTO {
session_id: String
session_id: String,
}
#[derive(Debug, Serialize, Deserialize)]
struct SessionJoinDto {
session_id: String
session_id: String,
}
#[derive(Debug, Serialize)]
@@ -383,12 +474,14 @@ struct SessionCreatedDto {
#[derive(Debug, Serialize)]
struct SessionClosedDto {
session: Session,
reason: String
reason: String,
}
#[derive(Debug, Clone, PartialEq)]
enum WebSocketConnectionType {
HOST, PEER, OBSERVER
HOST,
PEER,
OBSERVER,
}
impl FromStr for WebSocketConnectionType {
@@ -399,7 +492,7 @@ impl FromStr for WebSocketConnectionType {
"host" => Ok(WebSocketConnectionType::HOST),
"peer" => Ok(WebSocketConnectionType::PEER),
"observer" => Ok(WebSocketConnectionType::OBSERVER),
_ => Err(())
_ => Err(()),
}
}
}
@@ -407,14 +500,16 @@ impl FromStr for WebSocketConnectionType {
#[derive(Debug, Clone)]
struct WebSocketSession {
pub connection_type: WebSocketConnectionType,
pub session: Session
pub session: Session,
}
impl WebSocketConnectionType {
pub fn get_topics(&self) -> &[&str] {
match self {
WebSocketConnectionType::HOST => &["input", "session"],
WebSocketConnectionType::PEER | WebSocketConnectionType::OBSERVER => &["move", "input", "status", "session"],
WebSocketConnectionType::PEER | WebSocketConnectionType::OBSERVER => {
&["move", "input", "status", "session"]
}
}
}
}

125
server/src/kafka.rs
View File

@@ -1,20 +1,20 @@
use crate::hash::Hasher;
use crate::session::Session;
use hyper::{Body, Client, Method, Request, Uri};
use kafka::client::metadata::Topic;
use kafka::client::{KafkaClient, ProduceMessage};
use kafka::consumer::{Consumer, FetchOffset, GroupOffsetStorage, MessageSet};
use kafka::producer::{DefaultPartitioner, Partitioner, Producer, Record, RequiredAcks, Topics};
use pong::event::event::{Event, EventReaderImpl, EventWriter, EventWriterImpl};
use serde::Deserialize;
use std::hash::{BuildHasher, Hash};
use std::process::ExitStatus;
use std::str::FromStr;
use std::time::Duration;
use hyper::{Body, Client, Method, Request, Uri};
use serde::{Deserialize};
use kafka::client::{KafkaClient, ProduceMessage};
use kafka::client::metadata::Topic;
use tokio::process::Command;
use kafka::consumer::{Consumer, FetchOffset, GroupOffsetStorage, MessageSet};
use kafka::producer::{DefaultPartitioner, Partitioner, Producer, Record, RequiredAcks, Topics};
use pong::event::event::{Event, EventReaderImpl, EventWriter, EventWriterImpl};
use crate::hash::Hasher;
use crate::session::Session;
pub struct KafkaSessionEventWriterImpl {
producer: Producer<SessionPartitioner>
producer: Producer<SessionPartitioner>,
}
impl KafkaSessionEventWriterImpl {
pub fn new(host: &str) -> KafkaSessionEventWriterImpl {
@@ -25,14 +25,12 @@ impl KafkaSessionEventWriterImpl {
.with_partitioner(SessionPartitioner {})
.create()
.unwrap();
KafkaSessionEventWriterImpl {
producer
}
KafkaSessionEventWriterImpl { producer }
}
}
pub struct KafkaDefaultEventWriterImpl {
producer: Producer
producer: Producer,
}
impl KafkaDefaultEventWriterImpl {
pub fn new(host: &str) -> KafkaDefaultEventWriterImpl {
@@ -42,13 +40,10 @@ impl KafkaDefaultEventWriterImpl {
.with_required_acks(RequiredAcks::One)
.create()
.unwrap();
KafkaDefaultEventWriterImpl {
producer
}
KafkaDefaultEventWriterImpl { producer }
}
}
impl EventWriterImpl for KafkaSessionEventWriterImpl {
fn write(&mut self, event: Event) -> Result<(), String> {
match event.key {
@@ -56,14 +51,14 @@ impl EventWriterImpl for KafkaSessionEventWriterImpl {
let record = Record::from_key_value(event.topic.as_str(), key, event.msg.as_str());
match self.producer.send(&record) {
Ok(()) => Ok(()),
Err(e) => Err(format!("{}", e))
Err(e) => Err(format!("{}", e)),
}
},
}
None => {
let record = Record::from_value(event.topic.as_str(), event.msg.as_str());
match self.producer.send(&record) {
Ok(()) => Ok(()),
Err(e) => Err(format!("{}", e))
Err(e) => Err(format!("{}", e)),
}
}
}
@@ -77,14 +72,14 @@ impl EventWriterImpl for KafkaDefaultEventWriterImpl {
let record = Record::from_key_value(event.topic.as_str(), key, event.msg.as_str());
match self.producer.send(&record) {
Ok(()) => Ok(()),
Err(e) => Err(format!("{}", e))
Err(e) => Err(format!("{}", e)),
}
},
}
None => {
let record = Record::from_value(event.topic.as_str(), event.msg.as_str());
match self.producer.send(&record) {
Ok(()) => Ok(()),
Err(e) => Err(format!("{}", e))
Err(e) => Err(format!("{}", e)),
}
}
}
@@ -92,7 +87,7 @@ impl EventWriterImpl for KafkaDefaultEventWriterImpl {
}
pub struct KafkaEventReaderImpl {
consumer: Consumer
consumer: Consumer,
}
impl KafkaEventReaderImpl {
pub fn default() -> KafkaEventReaderImpl {
@@ -105,7 +100,7 @@ impl KafkaEventReaderImpl {
pub fn new(host: &str) -> KafkaEventReaderImpl {
println!("Connecting consumer to kafka host: {}", host);
let mut consumer = Consumer::from_hosts(vec!(host.to_owned()))
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())
@@ -114,14 +109,16 @@ impl KafkaEventReaderImpl {
.with_offset_storage(GroupOffsetStorage::Kafka)
.create()
.unwrap();
KafkaEventReaderImpl {
consumer
}
KafkaEventReaderImpl { consumer }
}
pub fn for_partitions(host: &str, partitions: &[i32], topics: &[&str]) -> Result<KafkaEventReaderImpl, String> {
pub fn for_partitions(
host: &str,
partitions: &[i32],
topics: &[&str],
) -> Result<KafkaEventReaderImpl, String> {
println!("Connecting partition specific consumer to kafka host {} with topics {:?} / partitions {:?}", host, topics, partitions);
let mut builder = Consumer::from_hosts(vec!(host.to_owned()));
let mut builder = Consumer::from_hosts(vec![host.to_owned()]);
for topic in topics.iter() {
builder = builder.with_topic_partitions(topic.parse().unwrap(), partitions);
}
@@ -130,16 +127,13 @@ impl KafkaEventReaderImpl {
.with_group("group".to_owned())
.with_offset_storage(GroupOffsetStorage::Kafka);
let consumer = builder
.create();
let consumer = builder.create();
if let Err(e) = consumer {
eprintln!("Failed to connect consumer: {:?}", e);
return Err("Failed to connect consumer".to_string());
}
let consumer = consumer.unwrap();
Ok(KafkaEventReaderImpl {
consumer
})
Ok(KafkaEventReaderImpl { consumer })
}
}
impl EventReaderImpl for KafkaEventReaderImpl {
@@ -162,11 +156,18 @@ impl KafkaEventReaderImpl {
let partition = ms.partition();
println!("querying topic={} partition={}", topic, partition);
for m in ms.messages() {
let event = Event {topic: String::from(topic), key: Some(std::str::from_utf8(m.key).unwrap().parse().unwrap()), msg: std::str::from_utf8(m.value).unwrap().parse().unwrap() };
let event = Event {
topic: String::from(topic),
key: Some(std::str::from_utf8(m.key).unwrap().parse().unwrap()),
msg: std::str::from_utf8(m.value).unwrap().parse().unwrap(),
};
topic_event_count += 1;
events.push(event);
}
println!("returned {:?} events for topic={} partition={}", topic_event_count, topic, partition);
println!(
"returned {:?} events for topic={} partition={}",
topic_event_count, topic, partition
);
self.consumer.consume_messageset(ms).unwrap();
}
self.consumer.commit_consumed().unwrap();
@@ -175,20 +176,22 @@ impl KafkaEventReaderImpl {
}
pub struct KafkaSessionEventReaderImpl {
inner: KafkaEventReaderImpl
inner: KafkaEventReaderImpl,
}
impl KafkaSessionEventReaderImpl {
pub fn new(host: &str, session: &Session, topics: &[&str]) -> Result<KafkaSessionEventReaderImpl, String> {
pub fn new(
host: &str,
session: &Session,
topics: &[&str],
) -> Result<KafkaSessionEventReaderImpl, String> {
let partitions = [session.id as i32];
let reader = KafkaEventReaderImpl::for_partitions(host, &partitions, topics);
if let Err(e) = reader {
return Err("Failed to create kafka session event reader".to_string());
}
let reader = reader.unwrap();
Ok(KafkaSessionEventReaderImpl {
inner: reader
})
Ok(KafkaSessionEventReaderImpl { inner: reader })
}
}
@@ -198,24 +201,31 @@ impl EventReaderImpl for KafkaSessionEventReaderImpl {
}
}
#[derive(Debug)]
pub struct KafkaTopicManager {
partition_management_endpoint: String
partition_management_endpoint: String,
}
impl KafkaTopicManager {
pub fn default() -> KafkaTopicManager {
KafkaTopicManager {partition_management_endpoint: "http://localhost: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()}
KafkaTopicManager {
partition_management_endpoint: format!("http://{}/add_partition", topic_manager_host)
.to_owned(),
}
}
pub async fn add_partition(&self) -> Result<u16, String> {
let mut client = Client::new();
let request = Request::builder().method(Method::POST).uri(Uri::from_str(&self.partition_management_endpoint).unwrap()).body(Body::empty()).unwrap();
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);
@@ -243,19 +253,26 @@ impl KafkaTopicManager {
}
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);
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);
println!(
"Successfully created partition: {}",
updated_partition_count
);
Ok(updated_partition_count)
}
}
#[derive(Deserialize)]
struct PartitionApiDTO {
data: u16
data: u16,
}
pub struct SessionPartitioner {}
@@ -267,8 +284,8 @@ impl Partitioner for SessionPartitioner {
let key = std::str::from_utf8(key).unwrap();
msg.partition = key.parse::<i32>().unwrap();
// println!("Overriding message partition with key: {}", msg.partition);
},
None => panic!("Producing message without key not allowed!")
}
None => panic!("Producing message without key not allowed!"),
}
}
}

11
server/src/main.rs
View File

@@ -2,14 +2,17 @@ extern crate core;
use crate::http::HttpServer;
mod hash;
pub mod http;
pub mod kafka;
pub mod utils;
mod hash;
mod session;
mod player;
mod session;
pub mod utils;
#[tokio::main]
pub async fn main() {
HttpServer::new([127, 0, 0, 1], 4000, "localhost:9093").run().await.expect("failed to run server");
HttpServer::new([127, 0, 0, 1], 4000, "localhost:9093")
.run()
.await
.expect("failed to run server");
}

4
server/src/player.rs
View File

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

152
server/src/session.rs
View File

@@ -1,21 +1,24 @@
use crate::hash::Hasher;
use crate::kafka::{
KafkaDefaultEventWriterImpl, KafkaEventReaderImpl, KafkaSessionEventReaderImpl,
KafkaSessionEventWriterImpl, KafkaTopicManager,
};
use crate::player::Player;
use kafka::producer::Producer;
use pong::event::event::{Event, EventReader, EventWriter};
use serde::de::DeserializeOwned;
use serde::{Deserialize, Serialize};
use serde_json::json;
use std::collections::HashMap;
use std::rc::Rc;
use std::sync::Arc;
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,
sessions: Vec<Session>,
session_producer: EventWriter,
topic_manager: KafkaTopicManager
topic_manager: KafkaTopicManager,
}
// TODO: On startup read the session events from kafka to restore the session id <-> hash mappings.
@@ -25,12 +28,17 @@ impl SessionManager {
kafka_host: kafka_host.to_owned(),
sessions: vec![],
topic_manager: KafkaTopicManager::from("localhost:7243"),
session_producer: EventWriter::new(Box::new(KafkaDefaultEventWriterImpl::new(kafka_host)))
session_producer: EventWriter::new(Box::new(KafkaDefaultEventWriterImpl::new(
kafka_host,
))),
}
}
pub fn get_session(&self, session_id: &str) -> Option<Session> {
self.sessions.iter().find(|s| s.hash == session_id).map_or_else(|| None, |s| Some(s.clone()))
self.sessions
.iter()
.find(|s| s.hash == session_id)
.map_or_else(|| None, |s| Some(s.clone()))
}
pub async fn create_session(&mut self, player: Player) -> Result<Session, String> {
@@ -41,14 +49,18 @@ impl SessionManager {
}
let session_id = add_partition_res.unwrap();
let session_hash = Hasher::hash(session_id);
let session = Session::new (session_id, session_hash, player.clone());
let session = Session::new(session_id, session_hash, player.clone());
println!("Successfully created session: {:?}", session);
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> {
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 {
@@ -64,6 +76,10 @@ impl SessionManager {
let error = format!("Can't join session with more than 1 player: {}", session_id);
return Err(error);
}
if session.players[0] == player {
let error = format!("Can't join session, because player {:?} is already in session: {}", player, session_id);
return Err(error);
}
session.players.push(player.clone());
session.state = SessionState::RUNNING;
session.clone()
@@ -75,9 +91,16 @@ impl SessionManager {
Ok(updated_session.clone())
}
fn write_to_producer<T>(&mut self, session_event: T) -> Result<(), String> where T : Serialize {
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});
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);
println!("{}", e);
@@ -87,47 +110,65 @@ impl SessionManager {
return Ok(());
}
pub fn split(&self, session_id: &str, read_topics: &[&str]) -> Result<(SessionReader, SessionWriter), String> {
pub fn split(
&self,
session_id: &str,
read_topics: &[&str],
) -> Result<(SessionReader, SessionWriter), String> {
let reader = self.get_session_reader(session_id, read_topics);
if let Err(e) = reader {
println!("Failed to create session reader: {:?}", e);
return Err("Failed to create session reader".to_string())
return Err("Failed to create session reader".to_string());
}
let writer = self.get_session_writer(session_id);
if let Err(e) = writer {
println!("Failed to create session writer: {:?}", e);
return Err("Failed to create session writer".to_string())
return Err("Failed to create session writer".to_string());
}
return Ok((reader.unwrap(), writer.unwrap()))
return Ok((reader.unwrap(), writer.unwrap()));
}
pub fn get_session_reader(&self, session_id: &str, topics: &[&str]) -> Result<SessionReader, String> {
pub fn get_session_reader(
&self,
session_id: &str,
topics: &[&str],
) -> Result<SessionReader, String> {
let session = self.find_session(&session_id);
if let None = session {
return Err(format!("Unable to find session with hash {}", session_id))
return Err(format!("Unable to find session with hash {}", session_id));
}
let session = session.unwrap();
let kafka_reader = KafkaSessionEventReaderImpl::new(&self.kafka_host, &session, topics);
if let Err(_) = kafka_reader {
return Err("Unable to create kafka reader.".to_string())
return Err("Unable to create kafka reader.".to_string());
}
let kafka_reader = kafka_reader.unwrap();
let event_reader = EventReader::new(Box::new(kafka_reader));
Ok(SessionReader {reader: event_reader, session})
Ok(SessionReader {
reader: event_reader,
session,
})
}
pub fn get_session_writer(&self, session_id: &str) -> Result<SessionWriter, String> {
let session = self.find_session(&session_id);
if let None = session {
return Err(format!("Unable to find session with hash {}", session_id))
return Err(format!("Unable to find session with hash {}", session_id));
}
let session = session.unwrap();
let event_writer = EventWriter::new(Box::new(KafkaSessionEventWriterImpl::new(&self.kafka_host)));
Ok(SessionWriter {writer: event_writer, session})
let event_writer =
EventWriter::new(Box::new(KafkaSessionEventWriterImpl::new(&self.kafka_host)));
Ok(SessionWriter {
writer: event_writer,
session,
})
}
fn find_session(&self, session_id: &str) -> Option<Session> {
self.sessions.iter().find(|s| session_id == s.hash).map(|s| s.clone())
self.sessions
.iter()
.find(|s| session_id == s.hash)
.map(|s| s.clone())
}
}
@@ -154,26 +195,39 @@ impl CachingSessionManager {
self.inner.create_session(player).await
}
pub async fn join_session(&mut self, session_id: String, player: Player) -> Result<Session, String> {
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> {
pub fn get_session_reader(
&mut self,
session_id: &str,
) -> Result<Arc<Mutex<SessionReader>>, String> {
let cached = self.reader_cache.get(session_id);
if let Some(reader) = cached {
println!("Reusing existing reader for session: {:?}", session_id);
return Ok(Arc::clone(reader));
}
let reader = self.inner.get_session_reader(session_id, &["move", "input", "status", "session"]);
let reader = self
.inner
.get_session_reader(session_id, &["move", "input", "status", "session"]);
if let Err(e) = reader {
return Err(e);
}
let reader = Arc::new(Mutex::new(reader.unwrap()));
self.reader_cache.insert(session_id.to_string(), Arc::clone(&reader));
self.reader_cache
.insert(session_id.to_string(), Arc::clone(&reader));
return Ok(Arc::clone(&reader));
}
pub fn get_session_writer(&mut self, session_id: &str) -> Result<Arc<Mutex<SessionWriter>>, String> {
pub fn get_session_writer(
&mut self,
session_id: &str,
) -> Result<Arc<Mutex<SessionWriter>>, String> {
let cached = self.writer_cache.get(session_id);
if let Some(writer) = cached {
println!("Reusing existing writer for session: {:?}", session_id);
@@ -184,7 +238,8 @@ impl CachingSessionManager {
return Err(e);
}
let writer = Arc::new(Mutex::new(writer.unwrap()));
self.writer_cache.insert(session_id.to_string(), Arc::clone(&writer));
self.writer_cache
.insert(session_id.to_string(), Arc::clone(&writer));
return Ok(Arc::clone(&writer));
}
}
@@ -194,16 +249,16 @@ pub struct Session {
pub id: u16,
pub hash: String,
pub state: SessionState,
players: Vec<Player>
players: Vec<Player>,
}
impl Session {
pub fn new(id: u16, hash: String, player: Player) -> Session {
Session {
players: vec!(player),
players: vec![player],
id,
hash,
state: SessionState::PENDING
state: SessionState::PENDING,
}
}
@@ -213,7 +268,7 @@ impl Session {
pub fn join(&mut self, player: Player) -> bool {
if !self.can_be_joined() {
return false
return false;
}
self.players.push(player);
return true;
@@ -224,24 +279,28 @@ impl Session {
pub enum SessionState {
PENDING, // 1 player is missing
RUNNING, // game is playing
CLOSED // game is over
CLOSED, // game is over
}
pub struct SessionWriter {
session: Session,
writer: EventWriter
writer: EventWriter,
}
impl SessionWriter {
pub fn write_to_session(&mut self, topic: &str, msg: &str) -> Result<(), String> {
let event = Event {msg: msg.to_owned(), key: Some(self.session.id.to_string()), topic: topic.to_owned()};
let event = Event {
msg: msg.to_owned(),
key: Some(self.session.id.to_string()),
topic: topic.to_owned(),
};
self.writer.write(event)
}
}
pub struct SessionReader {
session: Session,
reader: EventReader
reader: EventReader,
}
impl SessionReader {
@@ -254,7 +313,7 @@ impl SessionReader {
struct SessionCreatedEvent {
event_type: SessionEventType,
session: Session,
player: Player
player: Player,
}
impl SessionCreatedEvent {
@@ -262,7 +321,7 @@ impl SessionCreatedEvent {
SessionCreatedEvent {
event_type: SessionEventType::CREATED,
session,
player
player,
}
}
}
@@ -271,7 +330,7 @@ impl SessionCreatedEvent {
struct SessionJoinedEvent {
event_type: SessionEventType,
session: Session,
player: Player
player: Player,
}
impl SessionJoinedEvent {
@@ -279,7 +338,7 @@ impl SessionJoinedEvent {
SessionJoinedEvent {
event_type: SessionEventType::JOINED,
session,
player
player,
}
}
}
@@ -294,5 +353,6 @@ fn session_joined(session: Session, player: Player) -> SessionJoinedEvent {
#[derive(Deserialize, Serialize)]
enum SessionEventType {
CREATED, JOINED
CREATED,
JOINED,
}

38
server/src/utils.rs
View File

@@ -1,11 +1,11 @@
pub mod http_utils {
use hyper::body::Buf;
use hyper::{body, Body, Request};
use serde::de::DeserializeOwned;
use serde::Deserialize;
use std::borrow::BorrowMut;
use std::collections::HashMap;
use std::io::Read;
use hyper::{Body, body, Request};
use hyper::body::Buf;
use serde::de::DeserializeOwned;
use serde::Deserialize;
pub fn get_query_params(req: &Request<Body>) -> HashMap<&str, &str> {
let uri = req.uri();
@@ -13,13 +13,18 @@ pub mod http_utils {
println!("uri={:?}, query={:?}", uri, query);
match query {
None => HashMap::new(),
Some(query) => {
query.split("&").map(|s| s.split_at(s.find("=").unwrap())).map(|(key, value)| (key, &value[1..])).collect()
}
Some(query) => query
.split("&")
.map(|s| s.split_at(s.find("=").unwrap()))
.map(|(key, value)| (key, &value[1..]))
.collect(),
}
}
pub async fn read_json_body<T>(req: &mut Request<Body>) -> T where T : DeserializeOwned {
pub async fn read_json_body<T>(req: &mut Request<Body>) -> T
where
T: DeserializeOwned,
{
let mut body = req.body_mut();
let bytes = body::to_bytes(body).await.unwrap();
let body_str = std::str::from_utf8(&*bytes).unwrap();
@@ -29,12 +34,12 @@ pub mod http_utils {
#[cfg(test)]
pub mod http_utils_tests {
use super::*;
use crate::utils::http_utils::get_query_params;
use hyper::http::uri::{Builder, Parts};
use hyper::{Body, Request, Uri};
use rstest::rstest;
use std::collections::HashMap;
use hyper::{Body, Request, Uri};
use hyper::http::uri::{Builder, Parts};
use crate::utils::http_utils::get_query_params;
use super::*;
#[rstest]
#[case(
@@ -50,7 +55,12 @@ pub mod http_utils_tests {
HashMap::from([("topic", "status"), ("key", "abc")])
)]
fn get_query_params_tests(#[case] query_str: &str, #[case] expected: HashMap<&str, &str>) {
let uri = Builder::new().scheme("https").authority("behnke.rs").path_and_query(query_str).build().unwrap();
let uri = Builder::new()
.scheme("https")
.authority("behnke.rs")
.path_and_query(query_str)
.build()
.unwrap();
let req = Request::get(uri).body(Body::empty()).unwrap();
let res = get_query_params(&req);
assert_eq!(res, expected)
@@ -65,6 +75,6 @@ pub mod time_utils {
let since_the_epoch = start
.duration_since(UNIX_EPOCH)
.expect("Time went backwards");
return since_the_epoch.as_millis()
return since_the_epoch.as_millis();
}
}