diff --git a/pong/src/event.rs b/pong/src/event.rs
index c16a757..1ab498a 100644
--- a/pong/src/event.rs
+++ b/pong/src/event.rs
@@ -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()
         }
     }
diff --git a/pong/src/pong.rs b/pong/src/pong.rs
index f99e662..6de2036 100644
--- a/pong/src/pong.rs
+++ b/pong/src/pong.rs
@@ -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;
 
diff --git a/server/src/event.rs b/server/src/event.rs
index f977040..c1e44af 100644
--- a/server/src/event.rs
+++ b/server/src/event.rs
@@ -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(),
         })
     }
diff --git a/server/src/kafka.rs b/server/src/kafka.rs
index 550ccac..cdb900b 100644
--- a/server/src/kafka.rs
+++ b/server/src/kafka.rs
@@ -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()
     }
 }
diff --git a/server/src/session_manager.rs b/server/src/session_manager.rs
index dbf41c6..af81abb 100644
--- a/server/src/session_manager.rs
+++ b/server/src/session_manager.rs
@@ -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()
     }
 }
diff --git a/server/src/websocket_handler.rs b/server/src/websocket_handler.rs
index 10a815c..6f14d8a 100644
--- a/server/src/websocket_handler.rs
+++ b/server/src/websocket_handler.rs
@@ -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 {