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

don't double serialize event - only the payload is relevant

Browse Source
This commit is contained in:
Thilo Behnke
2022-06-26 14:01:30 +02:00
parent 080c754eeb
commit cb4a8aabdc
6 changed files with 42 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
pong/src/event.rs
View File

@@ -5,22 +5,21 @@ pub mod event {
use std::io::Write;
#[derive(Debug, Deserialize, Serialize)]
pub struct Event {
pub struct EventWrapper {
pub topic: String,
pub key: Option<String>,
#[serde(skip_serializing)]
pub msg: String,
pub event: String,
}
pub trait EventWriterImpl: Send + Sync {
fn write(&mut self, events: Vec<Event>) -> Result<(), String>;
fn write(&mut self, events: Vec<EventWrapper>) -> Result<(), String>;
}
pub struct FileEventWriterImpl {}
impl EventWriterImpl for FileEventWriterImpl {
fn write(&mut self, events: Vec<Event>) -> Result<(), String> {
fn write(&mut self, events: Vec<EventWrapper>) -> Result<(), String> {
let event_buffer = events.iter().fold(vec![], |mut acc, e| {
acc.push(e.msg.as_bytes());
acc.push(e.event.as_bytes());
acc
}).concat();
let options = OpenOptions::new()
@@ -41,7 +40,7 @@ pub mod event {
pub struct NoopEventWriterImpl {}
impl EventWriterImpl for NoopEventWriterImpl {
fn write(&mut self, _events: Vec<Event>) -> Result<(), String> {
fn write(&mut self, _events: Vec<EventWrapper>) -> Result<(), String> {
Ok(())
}
}
@@ -67,17 +66,17 @@ pub mod event {
}
}
pub fn write(&mut self, event: Event) -> Result<(), String> {
pub fn write(&mut self, event: EventWrapper) -> Result<(), String> {
self.write_all(vec![event])
}
pub fn write_all(&mut self, events: Vec<Event>) -> Result<(), String> {
pub fn write_all(&mut self, events: Vec<EventWrapper>) -> Result<(), String> {
self.writer_impl.write(events)
}
}
pub trait EventReaderImpl: Send + Sync {
fn read(&mut self) -> Result<Vec<Event>, String>;
fn read(&mut self) -> Result<Vec<EventWrapper>, String>;
}
pub struct EventReader {
@@ -89,7 +88,7 @@ pub mod event {
EventReader { reader_impl }
}
pub fn read(&mut self) -> Result<Vec<Event>, String> {
pub fn read(&mut self) -> Result<Vec<EventWrapper>, String> {
self.reader_impl.read()
}
}

2
pong/src/pong.rs
View File

@@ -158,7 +158,7 @@ pub mod pong_collisions {
}
pub mod pong_events {
use crate::event::event::{Event, EventWriter};
use crate::event::event::{EventWrapper, EventWriter};
use crate::geom::vector::Vector;
use serde::Serialize;

4
server/src/event.rs
View File

@@ -140,10 +140,10 @@ mod tests {
id: 1,
session_id: "abc".to_owned(),
state: SessionState::PENDING,
players: vec![Player { id: "player_1".to_owned() }],
players: vec![Player { id: "player_1".to_owned(), nr: 1, ip: "127.0.0.1".to_owned() }],
observers: vec![Observer {id: "observer_1".to_owned()}]
},
actor: Player { id: "player_1".to_owned() },
actor: Player { id: "player_1".to_owned(), nr: 1, ip: "127.0.0.1".to_owned() },
reason: "some reason".to_owned(),
})
}

22
server/src/kafka.rs
View File

@@ -9,7 +9,7 @@ use kafka::producer::{Partitioner, Producer, Record, RequiredAcks, Topics};
use log::{debug, error, info, trace};
use serde::Deserialize;
use pong::event::event::{Event, EventReaderImpl, EventWriterImpl};
use pong::event::event::{EventWrapper, EventReaderImpl, EventWriterImpl};
use crate::session::Session;
pub struct KafkaSessionEventWriterImpl {
@@ -53,28 +53,28 @@ impl KafkaDefaultEventWriterImpl {
}
impl EventWriterImpl for KafkaSessionEventWriterImpl {
fn write(&mut self, events: Vec<Event>) -> Result<(), String> {
fn write(&mut self, events: Vec<EventWrapper>) -> Result<(), String> {
write_events(events, &mut self.producer)
}
}
impl EventWriterImpl for KafkaDefaultEventWriterImpl {
fn write(&mut self, events: Vec<Event>) -> Result<(), String> {
fn write(&mut self, events: Vec<EventWrapper>) -> Result<(), String> {
write_events(events, &mut self.producer)
}
}
fn write_events<T>(events: Vec<Event>, producer: &mut Producer<T>) -> Result<(), String> where T : Partitioner {
fn write_events<T>(events: Vec<EventWrapper>, producer: &mut Producer<T>) -> Result<(), String> where T : Partitioner {
let mut records_without_key = vec![];
let mut records_with_key = vec![];
for event in events.iter() {
match &event.key {
Some(key) => {
let record = Record::from_key_value(&event.topic, key.clone(), event.msg.clone());
let record = Record::from_key_value(&event.topic, key.clone(), event.event.clone());
records_with_key.push(record);
}
None => {
let record = Record::from_value(&event.topic, event.msg.clone());
let record = Record::from_value(&event.topic, event.event.clone());
records_without_key.push(record);
}
}
@@ -128,13 +128,13 @@ impl KafkaEventReaderImpl {
}
impl EventReaderImpl for KafkaEventReaderImpl {
fn read(&mut self) -> Result<Vec<Event>, String> {
fn read(&mut self) -> Result<Vec<EventWrapper>, String> {
self.consume()
}
}
impl KafkaEventReaderImpl {
fn consume(&mut self) -> Result<Vec<Event>, String> {
fn consume(&mut self) -> Result<Vec<EventWrapper>, String> {
debug!("kafka consumer called to consume messages for {:?} / {:?}", self.topics, self.partitions);
let polled = self.consumer.poll().unwrap();
let message_sets: Vec<MessageSet<'_>> = polled.iter().collect();
@@ -145,10 +145,10 @@ impl KafkaEventReaderImpl {
let partition = ms.partition();
trace!("querying kafka topic={} partition={}", topic, partition);
for m in ms.messages() {
let event = Event {
let event = EventWrapper {
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(),
event: std::str::from_utf8(m.value).unwrap().parse().unwrap(),
};
topic_event_count += 1;
events.push(event);
@@ -186,7 +186,7 @@ impl KafkaSessionEventReaderImpl {
}
impl EventReaderImpl for KafkaSessionEventReaderImpl {
fn read(&mut self) -> Result<Vec<Event>, String> {
fn read(&mut self) -> Result<Vec<EventWrapper>, String> {
self.inner.read()
}
}

26
server/src/session_manager.rs
View File

@@ -2,7 +2,7 @@ use std::collections::HashMap;
use log::{debug, error, info};
use serde::{Deserialize, Serialize};
use pong::event::event::{Event, EventReader, EventWriter};
use pong::event::event::{EventWrapper, EventReader, EventWriter};
use crate::hash::Hasher;
use crate::kafka::{
@@ -114,7 +114,6 @@ impl SessionManager {
where
T: SessionEvent,
{
let json_event = serde_json::to_string(&session_event).unwrap();
let session_id = session_event.session_id();
let session_producer = match self.session_producers.get_mut(session_id) {
Some(p) => p,
@@ -124,19 +123,22 @@ impl SessionManager {
self.session_producers.get_mut(session_id).expect("failed to retrieve newly created session writer")
}
};
let session_event = Event {
topic: "session".to_owned(),
key: None,
msg: json_event,
};
let json_event = serde_json::to_string(&session_event);
if let Err(e) = json_event {
let error = format!("failed to serialize session event: {}", e);
error!("{}", error);
return Err(error);
}
let json_event = json_event.unwrap();
info!("preparing to write session event to kafka: {}", json_event);
let serialized_event = serde_json::to_string(&session_event).expect("failed to serialize session event");
let session_event_write = session_producer.write_to_session("session", vec![&serialized_event]);
let session_event_write = session_producer.write_to_session("session", vec![&json_event]);
if let Err(e) = session_event_write {
let message = format!("Failed to write session event to kafka: {:?}", e);
println!("{}", e);
return Err(message.to_owned());
}
info!("successfully produced session event.");
info!("successfully produced session event: {:?}", serialized_event);
return Ok(());
}
@@ -210,8 +212,8 @@ pub struct SessionWriter {
impl SessionWriter {
pub fn write_to_session(&mut self, topic: &str, messages: Vec<&str>) -> Result<(), String> {
let events = messages.into_iter().map(|e| {
Event {
msg: e.to_owned(),
EventWrapper {
event: e.to_owned(),
key: Some(self.session.id.to_string()),
topic: topic.to_owned(),
}
@@ -227,7 +229,7 @@ pub struct SessionReader {
}
impl SessionReader {
pub fn read_from_session(&mut self) -> Result<Vec<Event>, String> {
pub fn read_from_session(&mut self) -> Result<Vec<EventWrapper>, String> {
self.reader.read()
}
}

8
server/src/websocket_handler.rs
View File

@@ -183,13 +183,13 @@ impl WebsocketHandler for DefaultWebsocketHandler {
);
loop {
trace(&websocket_session_write_copy, "reading messages from kafka");
let messages = event_reader.read_from_session();
if let Err(e) = messages {
let events = event_reader.read_from_session();
if let Err(e) = events {
error(&websocket_session_write_copy, &format!("Failed to read messages from kafka: {:?}", e));
continue;
}
trace(&websocket_session_write_copy, &format!("read messages for websocket_session from consumer: {:?}", messages));
let messages = messages.unwrap();
trace(&websocket_session_write_copy, &format!("read messages for websocket_session from consumer: {:?}", events));
let messages = events.unwrap().into_iter().map(|e| e.event).collect::<Vec<String>>();
if messages.len() == 0 {
trace(&websocket_session_write_copy, "no new messages from kafka.");
} else {