.

Author: yoshuawuyts

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "async-sse"
-version = "1.0.0"
+version = "1.0.1"
 license = "MIT OR Apache-2.0"
 repository = "https://github.com/{{USERNAME}}/async-sse"
 documentation = "https://docs.rs/async-sse"
@@ -16,14 +16,15 @@ authors = [
 [features]
 
 [dependencies]
+async-std = "1.5.0"
 http-types = "1.0.1"
-async-std = { version = "1.5.0", features = ["attributes"] }
 log = "0.4.8"
-pin-project-lite = "0.1.4"
 memchr = "2.3.3"
+pin-project-lite = "0.1.4"
 
 [dev-dependencies]
 femme = "1.3.0"
+async-std = { version = "1.5.0", features = ["attributes"] }
 
 [patch.crates-io]
 async-std = { path = "../async-std" }

src/encoder.rs

@@ -1,25 +1,71 @@
+use async_std::io::prelude::*;
 use async_std::io::Write as AsyncWrite;
 use std::io;
 use std::time::Duration;
 
-use crate::Message;
-
 /// An SSE protocol encoder.
 #[derive(Debug)]
-pub struct Encoder;
+pub struct Encoder<W> {
+    writer: W,
+}
 
 /// Encode a new SSE connection.
-pub fn encode(_s: impl AsyncWrite) -> Encoder {
-    todo!();
+pub fn encode<W: AsyncWrite + Unpin>(writer: W) -> Encoder<W> {
+    Encoder { writer }
 }
 
-impl Encoder {
+impl<W> Encoder<W> {
+    /// Access the inner writer from the Encoder.
+    pub fn into_writer(self) -> W {
+        self.writer
+    }
+}
+
+impl<W: AsyncWrite + Unpin> Encoder<W> {
     /// Send a new message over SSE.
-    pub fn send(&self, _msg: Message) -> io::Result<()> {
-        todo!();
+    pub async fn send(&mut self, name: &str, data: &[u8], id: Option<&str>) -> io::Result<()> {
+        // Write the event name
+        self.writer.write_all(b"event:").await?;
+        self.writer.write_all(name.as_bytes()).await?;
+        self.writer.write_all(b"\n").await?;
+
+        // Write the id
+        if let Some(id) = id {
+            self.writer.write_all(b"id:").await?;
+            self.writer.write_all(id.as_bytes()).await?;
+            self.writer.write_all(b"\n").await?;
+        }
+
+        // Write the section
+        self.writer.write_all(b"data:").await?;
+        self.writer.write_all(data).await?;
+        self.writer.write_all(b"\n").await?;
+
+        // Finalize the message
+        self.writer.write_all(b"\n").await?;
+
+        Ok(())
     }
+
     /// Send a new "retry" message over SSE.
-    pub fn send_retry(&self, _dur: Duration) -> io::Result<()> {
-        todo!();
+    pub async fn send_retry(&mut self, dur: Duration, id: Option<&str>) -> io::Result<()> {
+        // Write the id
+        if let Some(id) = id {
+            self.writer.write_all(b"id:").await?;
+            self.writer.write_all(id.as_bytes()).await?;
+            self.writer.write_all(b"\n").await?;
+        }
+
+        // Write the section
+        self.writer.write_all(b"retry:").await?;
+        self.writer
+            .write_all(&format!("{}", dur.as_secs_f64() as u64).as_bytes())
+            .await?;
+        self.writer.write_all(b"\n").await?;
+
+        // Finalize the message
+        self.writer.write_all(b"\n").await?;
+
+        Ok(())
     }
 }

src/event.rs

@@ -2,7 +2,7 @@ use crate::Message;
 
 use std::time::Duration;
 
-/// Event kind.
+/// The kind of event sent.
 #[derive(Debug, Eq, PartialEq)]
 pub enum Event {
     /// A retry frame, signaling a new retry duration must be used..

src/lib.rs

@@ -1,6 +1,29 @@
 //! Async Server Sent Event parser and encoder.
 //!
-//! ## References
+//! # Example
+//!
+//! ```
+//! use async_sse::{decode, encode, Event};
+//! use async_std::io::Cursor;
+//! use async_std::prelude::*;
+//!
+//! #[async_std::main]
+//! async fn main() -> http_types::Result<()> {
+//!     let buf = Cursor::new(vec![]);
+//!     let mut encoder = encode(buf);
+//!     encoder.send("cat", b"chashu", None).await?;
+//!
+//!     let mut buf = encoder.into_writer();
+//!     buf.set_position(0);
+//!
+//!     let mut reader = decode(buf);
+//!     let event: Event = reader.next().await.unwrap()?;
+//!     # let _ = event;
+//!     Ok(())
+//! }
+//! ```
+//!
+//! # References
 //!
 //! - [SSE Spec](https://html.spec.whatwg.org/multipage/server-sent-events.html#concept-event-stream-last-event-id)
 //! - [EventSource web platform tests](https://github.com/web-platform-tests/wpt/tree/master/eventsource)

src/message.rs

@@ -1,4 +1,4 @@
-/// An SSE message.
+/// A data event.
 #[derive(Debug, PartialEq, Eq, Hash)]
 pub struct Message {
     /// The ID of this event.
@@ -26,4 +26,9 @@ impl Message {
     pub fn data(&self) -> &[u8] {
         &self.data
     }
+
+    /// Convert the message into the data payload.
+    pub fn into_bytes(self) -> Vec<u8> {
+        self.data
+    }
 }

tests/encode.rs

@@ -0,0 +1,67 @@
+use async_sse::{decode, encode, Event};
+use async_std::io::Cursor;
+use async_std::prelude::*;
+use std::time::Duration;
+
+/// Assert a Message.
+fn assert_message(event: &Event, name: &str, data: &str, id: Option<&'static str>) {
+    assert!(event.is_message());
+    if let Event::Message(msg) = event {
+        assert_eq!(msg.id(), &id.map(|s| s.to_owned()));
+        assert_eq!(msg.name(), name);
+        assert_eq!(
+            String::from_utf8(msg.data().to_owned()).unwrap(),
+            String::from_utf8(data.as_bytes().to_owned()).unwrap()
+        );
+    }
+}
+
+/// Assert a Message.
+fn assert_retry(event: &Event, dur: u64) {
+    assert!(event.is_retry());
+    let expected = Duration::from_secs_f64(dur as f64);
+    if let Event::Retry(dur) = event {
+        assert_eq!(dur, &expected);
+    }
+}
+#[async_std::test]
+async fn encode_message() -> http_types::Result<()> {
+    let buf = Cursor::new(vec![]);
+    let mut encoder = encode(buf);
+    encoder.send("cat", b"chashu", None).await?;
+    let mut buf = encoder.into_writer();
+    buf.set_position(0);
+
+    let mut reader = decode(buf);
+    let event = reader.next().await.unwrap()?;
+    assert_message(&event, "cat", "chashu", None);
+    Ok(())
+}
+
+#[async_std::test]
+async fn encode_message_with_id() -> http_types::Result<()> {
+    let buf = Cursor::new(vec![]);
+    let mut encoder = encode(buf);
+    encoder.send("cat", b"chashu", Some("0")).await?;
+    let mut buf = encoder.into_writer();
+    buf.set_position(0);
+
+    let mut reader = decode(buf);
+    let event = reader.next().await.unwrap()?;
+    assert_message(&event, "cat", "chashu", Some("0"));
+    Ok(())
+}
+
+#[async_std::test]
+async fn encode_retry() -> http_types::Result<()> {
+    let buf = Cursor::new(vec![]);
+    let mut encoder = encode(buf);
+    encoder.send_retry(Duration::from_secs(12), None).await?;
+    let mut buf = encoder.into_writer();
+    buf.set_position(0);
+
+    let mut reader = decode(buf);
+    let event = reader.next().await.unwrap()?;
+    assert_retry(&event, 12);
+    Ok(())
+}