Merge pull request #12 from Trivernis/develop

Event Data traits

Cargo.lock

@@ -148,7 +148,7 @@ dependencies = [
 
 [[package]]
 name = "rmp-ipc"
-version = "0.6.0"
+version = "0.6.1"
 dependencies = [
  "lazy_static",
  "log",

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "rmp-ipc"
-version = "0.6.0"
+version = "0.6.1"
 authors = ["trivernis <trivernis@protonmail.com>"]
 edition = "2018"
 readme = "README.md"

src/events/event.rs

@@ -1,6 +1,6 @@
 use crate::error::Result;
 use crate::events::generate_event_id;
-use serde::de::DeserializeOwned;
+use crate::events::payload::EventReceivePayload;
 use serde::{Deserialize, Serialize};
 use tokio::io::{AsyncRead, AsyncReadExt};
 
@@ -45,8 +45,8 @@ impl Event {
     }
 
     /// Decodes the data to the given type
-    pub fn data<T: DeserializeOwned>(&self) -> Result<T> {
-        let data = rmp_serde::from_read(&self.data[..])?;
+    pub fn data<T: EventReceivePayload>(&self) -> Result<T> {
+        let data = T::from_payload_bytes(&self.data[..])?;
 
         Ok(data)
     }

src/events/mod.rs

@@ -5,6 +5,7 @@ use std::time::{SystemTime, UNIX_EPOCH};
 pub mod error_event;
 pub mod event;
 pub mod event_handler;
+pub mod payload;
 
 /// Generates a new event id
 pub(crate) fn generate_event_id() -> u64 {

src/events/payload.rs

@@ -0,0 +1,65 @@
+use crate::prelude::IPCResult;
+use serde::de::DeserializeOwned;
+use serde::Serialize;
+use std::io::Read;
+
+/// Trait to convert event data into sending bytes
+/// It is implemented for all types that implement Serialize
+pub trait EventSendPayload {
+    fn to_payload_bytes(self) -> IPCResult<Vec<u8>>;
+}
+
+impl<T> EventSendPayload for T
+where
+    T: Serialize,
+{
+    fn to_payload_bytes(self) -> IPCResult<Vec<u8>> {
+        let bytes = rmp_serde::to_vec(&self)?;
+
+        Ok(bytes)
+    }
+}
+
+/// Trait to get the event data from receiving bytes.
+/// It is implemented for all types that are DeserializeOwned
+pub trait EventReceivePayload: Sized {
+    fn from_payload_bytes<R: Read>(reader: R) -> IPCResult<Self>;
+}
+
+impl<T> EventReceivePayload for T
+where
+    T: DeserializeOwned,
+{
+    fn from_payload_bytes<R: Read>(reader: R) -> IPCResult<Self> {
+        let type_data = rmp_serde::from_read(reader)?;
+        Ok(type_data)
+    }
+}
+
+/// A payload wrapper type for sending bytes directly without
+/// serializing them
+#[derive(Clone, Debug)]
+pub struct BytePayload {
+    bytes: Vec<u8>,
+}
+
+impl BytePayload {
+    pub fn new(bytes: Vec<u8>) -> Self {
+        Self { bytes }
+    }
+}
+
+impl EventSendPayload for BytePayload {
+    fn to_payload_bytes(self) -> IPCResult<Vec<u8>> {
+        Ok(self.bytes)
+    }
+}
+
+impl EventReceivePayload for BytePayload {
+    fn from_payload_bytes<R: Read>(mut reader: R) -> IPCResult<Self> {
+        let mut buf = Vec::new();
+        reader.read_to_end(&mut buf)?;
+
+        Ok(Self::new(buf))
+    }
+}

src/ipc/stream_emitter.rs

@@ -1,11 +1,12 @@
 use crate::error::Result;
 use crate::events::event::Event;
+use crate::events::payload::EventSendPayload;
 use crate::ipc::context::Context;
-use serde::Serialize;
 use std::sync::Arc;
 use tokio::io::AsyncWriteExt;
 use tokio::net::tcp::OwnedWriteHalf;
 use tokio::sync::Mutex;
+use tokio::time::Instant;
 
 /// An abstraction over the raw tokio tcp stream
 /// to emit events and share a connection across multiple
@@ -22,14 +23,14 @@ impl StreamEmitter {
         }
     }
 
-    pub async fn _emit<T: Serialize>(
+    pub async fn _emit<T: EventSendPayload>(
         &self,
         namespace: Option<&str>,
         event: &str,
         data: T,
         res_id: Option<u64>,
     ) -> Result<EmitMetadata> {
-        let data_bytes = rmp_serde::to_vec(&data)?;
+        let data_bytes = data.to_payload_bytes()?;
         log::debug!("Emitting event {:?}:{}", namespace, event);
 
         let event = if let Some(namespace) = namespace {
@@ -40,16 +41,17 @@ impl StreamEmitter {
 
         let event_bytes = event.to_bytes()?;
         {
-            log::trace!("Writing {} bytes", event_bytes.len());
+            let start = Instant::now();
             let mut stream = self.stream.lock().await;
             (*stream).write_all(&event_bytes[..]).await?;
+            log::trace!("Wrote {} bytes in {:?}", event_bytes.len(), start.elapsed());
         }
 
         Ok(EmitMetadata::new(event.id()))
     }
 
     /// Emits an event
-    pub async fn emit<S: AsRef<str>, T: Serialize>(
+    pub async fn emit<S: AsRef<str>, T: EventSendPayload>(
         &self,
         event: S,
         data: T,
@@ -58,7 +60,7 @@ impl StreamEmitter {
     }
 
     /// Emits an event to a specific namespace
-    pub async fn emit_to<S1: AsRef<str>, S2: AsRef<str>, T: Serialize>(
+    pub async fn emit_to<S1: AsRef<str>, S2: AsRef<str>, T: EventSendPayload>(
         &self,
         namespace: S1,
         event: S2,
@@ -69,7 +71,7 @@ impl StreamEmitter {
     }
 
     /// Emits a response to an event
-    pub async fn emit_response<S: AsRef<str>, T: Serialize>(
+    pub async fn emit_response<S: AsRef<str>, T: EventSendPayload>(
         &self,
         event_id: u64,
         event: S,
@@ -79,7 +81,7 @@ impl StreamEmitter {
     }
 
     /// Emits a response to an event to a namespace
-    pub async fn emit_response_to<S1: AsRef<str>, S2: AsRef<str>, T: Serialize>(
+    pub async fn emit_response_to<S1: AsRef<str>, S2: AsRef<str>, T: EventSendPayload>(
         &self,
         event_id: u64,
         namespace: S1,

src/lib.rs

@@ -109,6 +109,7 @@ mod namespaces;
 pub use events::error_event;
 pub use events::event;
 pub use events::event_handler;
+pub use events::payload;
 pub use ipc::builder::IPCBuilder;
 pub use macros::*;
 pub use namespaces::builder::NamespaceBuilder;
@@ -126,5 +127,6 @@ pub mod prelude {
     pub use crate::namespace::Namespace;
     pub use crate::namespaces::builder::NamespaceBuilder;
     pub use crate::namespaces::provider_trait::*;
+    pub use crate::payload::*;
     pub use crate::*;
 }

src/tests/ipc_tests.rs

@@ -1,4 +1,4 @@
-use self::super::utils::PingEventData;
+use super::utils::PingEventData;
 use crate::prelude::*;
 use crate::tests::utils::start_test_server;
 use std::sync::atomic::{AtomicBool, Ordering};