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

&str instead of String for event?

Browse Source
This commit is contained in:
Thilo Behnke
2022-06-19 00:58:14 +02:00
parent c245447750
commit f06b940c0a
3 changed files with 16 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
pong/src/event.rs
View File

@@ -5,10 +5,10 @@ pub mod event {
use std::io::Write;
#[derive(Debug, Deserialize, Serialize)]
pub struct Event {
pub topic: String,
pub key: Option<String>,
pub msg: String,
pub struct Event<'a> {
pub topic: &'a str,
pub key: Option<&'a str>,
pub msg: &'a str,
}
pub trait EventWriterImpl: Send + Sync {

14
server/src/kafka.rs
View File

@@ -55,23 +55,23 @@ impl EventWriterImpl for KafkaDefaultEventWriterImpl {
}
}
fn write_events<T>(events: Vec<Event>, producer: &mut Producer<T>) -> Result<(), String> where T : Partitioner {
fn write_events<'a, T>(events: Vec<Event<'a>>, 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 {
match &event.key {
Some(key) => {
let record = Record::from_key_value(event.topic.as_str(), key.clone(), event.msg.as_str());
let record = Record::from_key_value(event.topic, key.clone(), event.msg);
records_with_key.push(record);
}
None => {
let record = Record::from_value(event.topic.as_str(), event.msg.as_str());
let record = Record::from_value(event.topic, event.msg);
records_without_key.push(record);
}
}
}
let res_with_key = match producer.send_all::<String, &str>(&*records_with_key) {
let res_with_key = match producer.send_all::<&str, &str>(&*records_with_key) {
Ok(_) => Ok(()),
Err(e) => Err(format!("{}", e)),
};
@@ -155,9 +155,9 @@ impl KafkaEventReaderImpl {
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(),
topic,
key: Some(std::str::from_utf8(m.key).unwrap()),
msg: std::str::from_utf8(m.value).unwrap(),
};
topic_event_count += 1;
events.push(event);

10
server/src/session_manager.rs
View File

@@ -109,9 +109,9 @@ impl SessionManager {
{
let json_event = serde_json::to_string(&session_event).unwrap();
let session_event_write = self.session_producer.write(Event {
topic: "session".to_owned(),
topic: "session",
key: None,
msg: json_event,
msg: json_event.as_str(),
});
if let Err(e) = session_event_write {
let message = format!("Failed to write session create event to kafka: {:?}", e);
@@ -193,9 +193,9 @@ 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(),
key: Some(self.session.id.to_string()),
topic: topic.to_owned(),
msg: e,
key: Some(&self.session.id.to_string()),
topic,
}
}).collect();
self.writer.write_all(events)