Refactors

Cargo.lock

@@ -23,6 +23,7 @@ version = "0.1.0"
 dependencies = [
  "hd-lib",
  "tracing",
+ "tracing-subscriber",
 ]
 
 [[package]]

crates/hd-client/Cargo.toml

@@ -5,4 +5,4 @@ edition = "2024"
 
 [dependencies]
 tracing = { workspace = true }
-hd-lib = { path = "../hd-lib" }
+hd-lib = {path = "../hd-lib"}

crates/hd-client/src/bin/client.rs

@@ -1,7 +0,0 @@
-use hd_client::{EventBusClient, EventBusClientError};
-
-fn main() -> Result<(), EventBusClientError> {
-    let addr = "0.0.0.0:21368";
-    let mut client = EventBusClient::start(addr)?;
-    client.subscribe()
-}

crates/hd-client/src/lib.rs

@@ -1,56 +1,51 @@
-use std::{
+
-    io::{Read, Write},
+use std::net::TcpStream;
-    net::TcpStream,
+
+use hd_lib::{
+    bus::packet::{self, Packet},
+    error::HdError,
+    hd_tcp,
 };
+use tracing::{debug, info};
 
 #[derive(Debug)]
-pub enum EventBusClientError {
+pub enum HdClientError {
     TcpStreamError(std::io::Error),
+    HError(HdError),
 }
 
-pub struct EventBusClient {
+impl From<HdError> for HdClientError {
+    fn from(value: HdError) -> Self {
+        Self::HError(value)
+    }
+}
+
+pub struct HdClient {
     tcp_stream: TcpStream,
 }
 
-impl EventBusClient {
+impl HdClient {
-    pub fn start(addr: &'static str) -> Result<Self, EventBusClientError> {
+    pub fn start(addr: &'static str) -> Result<Self, HdClientError> {
-        let stream = TcpStream::connect(addr).map_err(EventBusClientError::TcpStreamError)?;
+        let stream = TcpStream::connect(addr).map_err(HdClientError::TcpStreamError)?;
-        println!("Event bus client connected to {}", addr);
+        info!("Event bus client connected to {}", addr);
-
         Ok(Self { tcp_stream: stream })
     }
 
-    pub fn subscribe(&mut self) -> Result<(), EventBusClientError> {
+    pub fn subscribe(&mut self) -> Result<(), HdClientError> {
-        let bytes = [1];
+        debug!("Going to subscribe");
-        self.tcp_stream
+        hd_tcp::write_packet(
-            .write_all(&bytes)
+            &mut self.tcp_stream,
-            .map_err(EventBusClientError::TcpStreamError)?;
+            &Packet::create_connect_packet(packet::ConnectMode::Subscribe),
-
+        )?;
-        self.tcp_stream.flush().map_err(EventBusClientError::TcpStreamError)?;
-
-        let mut buf = vec![0; 1024];
         loop {
-            let Ok(bytes_read) = self.tcp_stream.read(&mut buf) else {
+            let packet = hd_tcp::read_packet(&mut self.tcp_stream)?;
-                return Ok(());
+            if let Packet::Disconnect = packet {
-            };
+                info!("Disconnecting from the server");
-
-            if bytes_read == 0 {
-                eprintln!("Connection closed");
                 return Ok(());
             }
 
-            println!("MESSAGE BEGIN");
+            debug!("Received packet ... {}", packet::get_packet_type(&packet));
-            for b in &buf[0..bytes_read] {
+            hd_tcp::write_packet(&mut self.tcp_stream, &Packet::create_settle_packet(1))?;
-                print!("[{}] ", b);
-            }
-            println!("MESSAGE END");
-
-            // ACK
-            self.tcp_stream
-                .write_all(&bytes)
-                .map_err(EventBusClientError::TcpStreamError)?;
-
-            self.tcp_stream.flush().map_err(EventBusClientError::TcpStreamError)?;
         }
     }
 }

crates/hd-examples/Cargo.toml

@@ -0,0 +1,11 @@
+[package]
+name = "hd-examples"
+version = "0.1.0"
+edition = "2024"
+
+[dependencies]
+tracing = { workspace = true }
+hd-lib = { path = "../hd-lib" }
+hd-client = { path = "../hd-client" }
+hd-server = { path = "../hd-server" }
+tracing-subscriber = { version = "0.3", features = ["env-filter", "fmt"] }

crates/hd-examples/src/bin/client.rs

@@ -0,0 +1,11 @@
+use hd_client::{HdClient, HdClientError};
+
+fn main() -> Result<(), HdClientError> {
+    tracing_subscriber::fmt()
+        .with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
+        .init();
+
+    let addr = "0.0.0.0:21368";
+    let mut client = HdClient::start(addr)?;
+    client.subscribe()
+}

crates/hd-examples/src/bin/server.rs

@@ -1,23 +1,26 @@
 use std::{sync::Arc, thread, time::Duration};
 
-use hd_lib::{bus::EventData, error::HError};
+use hd_lib::{bus::EventData, error::HdError};
-use hd_server::TcpEventBus;
+use hd_server::HdBus;
 use tracing::info_span;
 
-fn main() -> Result<(), HError> {
+fn main() -> Result<(), HdError> {
     tracing_subscriber::fmt()
         .with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
         .init();
 
     let addr = "0.0.0.0:21368";
-    let _scope = info_span!("example_server").entered();
+    let server = Arc::new(HdBus::new());
-    let server = Arc::new(TcpEventBus::new());
     server.publish(EventData::new(1, "Hello"))?;
     server.publish(EventData::new(1, "World"))?;
 
     let server_clone = server.clone();
-    thread::spawn(move || server.start(addr));
+    thread::spawn(move || {
+        let _scope = info_span!("tcp_server").entered();
+        server.start(addr)
+    });
 
+    let _scope = info_span!("health_checks").entered();
     for i in 0..255 {
         thread::sleep(Duration::from_secs(5));
         server_clone.publish(EventData::new(i, "asd"))?;

crates/hd-lib/src/bus/mod.rs

@@ -1,8 +1,8 @@
-mod packet;
+pub mod packet;
 
-use tracing::{info, info_span};
+use tracing::{debug_span, info};
 
-use crate::error::HError;
+use crate::error::HdError;
 use std::{
     sync::{
         Arc, RwLock,
@@ -63,7 +63,12 @@ impl EventBusInner {
 struct SubscriptionHandle {
     id: SubscriptionId,
     new_event_signal: Sender<u64>,
-    join_handle: JoinHandle<Result<(), HError>>,
+    join_handle: JoinHandle<Result<(), HdError>>,
+}
+
+pub struct SubscriptionInfo {
+    pub id: SubscriptionId,
+    pub event_recv: Receiver<Arc<Event>>,
 }
 
 pub struct PollResult {
@@ -79,8 +84,8 @@ impl EventBus {
         }
     }
 
-    pub fn publish(&self, evt: EventData) -> Result<(), HError> {
+    pub fn publish(&self, evt: EventData) -> Result<(), HdError> {
-        let mut bus = self.inner.write().map_err(|e| HError::BusLockPoisoned(e.to_string()))?;
+        let mut bus = self.inner.write().map_err(|e| HdError::BusLockPoisoned(e.to_string()))?;
 
         let next_event_id = &mut bus.next_event_id;
         let event = Event {
@@ -95,6 +100,7 @@ impl EventBus {
 
         for s in &bus.subscription_handles {
             if let Err(e) = s.new_event_signal.send(1) {
+                // TODO: Multiple failures should automatically unsubscribe
                 tracing::error!("Failed to notify {}", e);
             }
         }
@@ -102,8 +108,8 @@ impl EventBus {
         Ok(())
     }
 
-    pub fn get_next_event(&self, cursor: Cursor) -> Result<Arc<Event>, HError> {
+    pub fn get_next_event(&self, cursor: Cursor) -> Result<Arc<Event>, HdError> {
-        let bus = self.inner.read().map_err(|e| HError::BusLockPoisoned(e.to_string()))?;
+        let bus = self.inner.read().map_err(|e| HdError::BusLockPoisoned(e.to_string()))?;
 
         for evt in &bus.events {
             if evt.id >= cursor {
@@ -111,22 +117,22 @@ impl EventBus {
             }
         }
 
-        Err(HError::NoMoreEvents)
+        Err(HdError::NoMoreEvents)
     }
 
-    pub fn subscribe(&self) -> Result<Receiver<Arc<Event>>, HError> {
+    pub fn subscribe(&self) -> Result<SubscriptionInfo, HdError> {
-        let mut bus = self.inner.write().map_err(|e| HError::BusLockPoisoned(e.to_string()))?;
+        let mut bus = self.inner.write().map_err(|e| HdError::BusLockPoisoned(e.to_string()))?;
 
         let next_subscription_id = bus.next_subscription_id;
-        // let next_event_id = bus.next_event_id;
         let next_event_id = 0;
         let (new_event_signal_sender, new_event_signal_recv) = channel();
         let (event_sender, event_recv) = sync_channel(10);
 
+        // Start event delivery in the background as and when new events are available
         let subscription = Subscription::new(next_subscription_id, next_event_id, new_event_signal_recv, event_sender);
-
         let bg_thread_handle = start_delivery(subscription, self.clone());
 
+        // Store a handle that the bus can use signal the bg delivery thread of new events
         let subscription_handle = SubscriptionHandle {
             id: next_subscription_id,
             new_event_signal: new_event_signal_sender,
@@ -136,11 +142,17 @@ impl EventBus {
         bus.next_subscription_id += 1;
         bus.subscription_handles.push(subscription_handle);
 
-        Ok(event_recv)
+        // Return info to caller that can be used to receive the events and unsubscribe later.
+        let subscription_info = SubscriptionInfo {
+            id: next_subscription_id,
+            event_recv,
+        };
+
+        Ok(subscription_info)
     }
 
-    pub fn unsubscribe(&self, id: SubscriptionId) -> Result<(), HError> {
+    pub fn unsubscribe(&self, id: SubscriptionId) -> Result<(), HdError> {
-        let mut bus = self.inner.write().map_err(|e| HError::BusLockPoisoned(e.to_string()))?;
+        let mut bus = self.inner.write().map_err(|e| HdError::BusLockPoisoned(e.to_string()))?;
 
         let mut found = None;
         let mut index = 0;
@@ -154,7 +166,7 @@ impl EventBus {
         }
 
         match found {
-            None => Err(HError::SubscriptionNotFound(format!(
+            None => Err(HdError::SubscriptionNotFound(format!(
                 "Subscription with id {} is not found",
                 id
             ))),
@@ -192,10 +204,11 @@ impl Subscription {
     }
 }
 
-fn start_delivery(mut subscription: Subscription, bus: EventBus) -> JoinHandle<Result<(), HError>> {
+fn start_delivery(subscription: Subscription, bus: EventBus) -> JoinHandle<Result<(), HdError>> {
     thread::spawn(move || {
+        let mut subscription = subscription;
         loop {
-            let _scope = info_span!("subscription_delivery", subscription.id).entered();
+            let _scope = debug_span!("subscription_delivery", subscription.id).entered();
             loop {
                 if let Ok(evt) = bus.get_next_event(subscription.cursor) {
                     info!("Delivering event {} ", evt.id);

crates/hd-lib/src/bus/packet.rs

@@ -1,305 +1,320 @@
-use std::{
+use std::{fmt::Display, sync::Arc};
-    io::{Read, Write},
-    net::TcpStream,
-};
 
-use crate::bus::EventData;
+use crate::bus::{Event, EventData};
 
-enum PacketType {
+mod packet_constants {
+    pub(crate) const HEADER_SIZE: usize = 3;
+}
+
+#[derive(PartialEq)]
+pub enum PacketType {
     Connect,
-    Unsubscribe,
     Publish,
-    SendEvent,
+    Deliver,
     Settle,
-    Ack,
     Disconnect,
 }
 
-enum Packet {
+#[derive(PartialEq)]
+pub enum ConnectMode {
+    Publish,
+    Subscribe,
+}
+
+impl Display for PacketType {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        let to_write = match self {
+            PacketType::Connect => "CONNECT",
+            PacketType::Publish => "PUBLISH",
+            PacketType::Deliver => "DELIVER",
+            PacketType::Settle => "SETTLE",
+            PacketType::Disconnect => "DISCONNECT",
+        };
+
+        write!(f, "{}", to_write)
+    }
+}
+
+pub enum Packet {
     Connect(ConnectPacket),
-    Unsubscribe,
     Publish(PublishPacket),
-    SendEvent(SendEventPacket),
+    Deliver(DeliverPacket),
     Settle(SettlePacket),
-    Ack(AckPacket),
     Disconnect,
 }
 
-struct ConnectPacket {
+pub struct ConnectPacket {
-    cursor: u64,
+    pub mode: ConnectMode,
 }
 
-struct PublishPacket {
+pub struct PublishPacket {
-    event_type: u8,
+    pub event_type: u8,
-    data_utf8: Vec<u8>,
+    pub data_utf8: Vec<u8>,
 }
 
-struct SendEventPacket {
+pub struct DeliverPacket {
     event_id: u64,
     event_type: u8,
     data_utf8: Vec<u8>,
 }
 
-struct SettlePacket {
+pub struct SettlePacket {
     cursor: u64,
 }
 
-struct AckPacket {
+pub struct AckPacket {
-    packet_type: PacketType,
+    pub packet_type: PacketType,
 }
 
 impl Packet {
-    fn get_packet_type(&self) -> PacketType {
+    pub fn create_connect_packet(mode: ConnectMode) -> Packet {
-        match self {
+        Packet::Connect(ConnectPacket { mode })
-            Packet::Connect(_) => PacketType::Connect,
-            Packet::Unsubscribe => PacketType::Unsubscribe,
-            Packet::Publish(_) => PacketType::Publish,
-            Packet::SendEvent(_) => PacketType::SendEvent,
-            Packet::Settle(_) => PacketType::Settle,
-            Packet::Ack(_) => PacketType::Ack,
-            Packet::Disconnect => PacketType::Disconnect,
-        }
     }
 
-    fn create_connect_packet(cursor: u64) -> Packet {
+    // TODO: Avoid copying data_utf8
-        Packet::Connect(ConnectPacket { cursor })
+    pub fn create_publish_packet(evt: EventData) -> Packet {
-    }
-
-    fn create_unsubscribe_packet() -> Packet {
-        Packet::Unsubscribe
-    }
-
-    fn create_publish_packet(evt: EventData) -> Packet {
         Packet::Publish(PublishPacket {
             event_type: evt.event_type,
             data_utf8: evt.data_utf8,
         })
     }
 
-    fn create_settle_packet(cursor: u64) -> Packet {
+    // TODO: Avoid copying data_utf8
+    pub fn create_deliver_packet(evt: Arc<Event>) -> Packet {
+        Packet::Deliver(DeliverPacket {
+            event_id: evt.id,
+            event_type: evt.data.event_type,
+            data_utf8: evt.data.data_utf8.clone(),
+        })
+    }
+
+    pub fn create_settle_packet(cursor: u64) -> Packet {
         Packet::Settle(SettlePacket { cursor })
     }
 
-    fn create_ack_packet(packet_type: PacketType) -> Packet {
+    pub fn create_disconnect_packet() -> Packet {
-        Packet::Ack(AckPacket { packet_type })
-    }
-
-    fn create_disconnect_packet() -> Packet {
         Packet::Disconnect
     }
 }
 
-struct PacketFrame {
+#[derive(Debug)]
-    header: HeaderFrame,
+pub enum PacketError {
-    data: DataFrame,
-}
-
-/// 3 Bytes
-struct HeaderFrame {
-    /// 0 Connect
-    /// 1 Subscribe
-    /// 2 Unsubscribe
-    /// 3 Publish
-    /// 4 Settle
-    /// 5 Ack
-    /// 6 Disconnect
-    packet_type: u8,
-    /// Length of data frame, Little Endian, so 5 = [5 0], 256 = [255 1]
-    data_frame_length: [u8; 2],
-}
-
-struct DataFrame {
-    data: Vec<u8>,
-}
-
-enum PacketError {
     WrongPacketType(u8),
     PacketProtocolError,
-    TcpStreamError(std::io::Error),
 }
 
-fn read_packet(stream: &mut TcpStream) -> Result<Packet, PacketError> {
+impl Display for PacketError {
-    let mut header_buf = vec![0; 3];
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-    stream
+        let to_write = match self {
-        .read_exact(&mut header_buf)
+            PacketError::WrongPacketType(p) => &format!("WRONG_PACKET_TYPE({})", p),
-        .map_err(PacketError::TcpStreamError)?;
+            PacketError::PacketProtocolError => "PACKET_PROTOCOL_ERROR",
+        };
 
-    let packet_type = get_packet_type_from_byte(header_buf[0])?;
+        write!(f, "{}", to_write)
+    }
+}
 
-    match packet_type {
+pub struct ReadPacketState {
+    pub required_buffer_size: usize,
+    packet_type: PacketType,
+}
+
+pub(crate) fn start_read_packet(buf: &mut [u8; packet_constants::HEADER_SIZE]) -> Result<ReadPacketState, PacketError> {
+    let packet_type = get_packet_type_from_byte(buf[0])?;
+    let required_buffer_size = read_u16(&buf[1..3])?;
+
+    Ok(ReadPacketState {
+        required_buffer_size,
+        packet_type,
+    })
+}
+
+pub(crate) fn read_packet(buf: &[u8], read_state: ReadPacketState) -> Result<Packet, PacketError> {
+    match read_state.packet_type {
         PacketType::Connect => {
-            let cursor = read_u64(stream)?;
+            let mode_byte = buf[0];
-            Ok(Packet::Connect(ConnectPacket { cursor }))
+            let mode = get_connect_mode_from_byte(mode_byte)?;
+            Ok(Packet::create_connect_packet(mode))
         }
-        PacketType::Unsubscribe => Ok(Packet::Unsubscribe),
         PacketType::Publish => {
-            let mut header = vec![0u8; 3];
+            let event_type = buf[0];
-            stream.read_exact(&mut header).map_err(PacketError::TcpStreamError)?;
+            let event_len = read_u16(&buf[1..3])?;
 
-            let event_type = header[0];
+            // TODO: Share underlying bytes without copying
-
+            let publish_packet = PublishPacket {
-            let len = usize::from_le_bytes([header[1], header[2], 0, 0, 0, 0, 0, 0]);
+                event_type,
-            let mut data_utf8 = vec![0u8; len];
+                data_utf8: buf[3..3 + event_len].to_vec(),
-            stream.read_exact(&mut data_utf8).map_err(PacketError::TcpStreamError)?;
+            };
-
-            let publish_packet = PublishPacket { event_type, data_utf8 };
             Ok(Packet::Publish(publish_packet))
         }
-        PacketType::SendEvent => {
+        PacketType::Deliver => {
-            let mut header = vec![0u8; 11];
+            let event_id = read_u64(&buf[0..8])?;
-            stream.read_exact(&mut header).map_err(PacketError::TcpStreamError)?;
+            let event_type = buf[8];
+            let event_len = read_u16(&buf[9..11])?;
 
-            let mut le_bytes = [0u8; 8];
+            let deliver_packet = DeliverPacket {
-            le_bytes.copy_from_slice(&header[0..8]);
-            let event_id = u64::from_le_bytes(le_bytes);
-
-            let event_type = header[8];
-
-            let len = usize::from_le_bytes([header[9], header[10], 0, 0, 0, 0, 0, 0]);
-            let mut data_utf8 = vec![0u8; len];
-            stream.read_exact(&mut data_utf8).map_err(PacketError::TcpStreamError)?;
-
-            let send_event_packet = SendEventPacket {
                 event_id,
                 event_type,
-                data_utf8,
+                data_utf8: buf[11..11 + event_len].to_vec(),
             };
-            Ok(Packet::SendEvent(send_event_packet))
+            Ok(Packet::Deliver(deliver_packet))
         }
         PacketType::Settle => {
-            let cursor = read_u64(stream)?;
+            let cursor = read_u64(&buf[0..8])?;
-            Ok(Packet::Settle(SettlePacket { cursor }))
+            Ok(Packet::create_settle_packet(cursor))
         }
-        PacketType::Ack => {
+        PacketType::Disconnect => Ok(Packet::create_disconnect_packet()),
-            let packet_type = get_packet_type_from_byte(read_byte(stream)?)?;
-            Ok(Packet::Ack(AckPacket { packet_type }))
-        }
-        PacketType::Disconnect => Ok(Packet::Disconnect),
     }
 }
 
-fn write_packet(stream: &mut TcpStream, packet: &Packet) -> Result<(), PacketError> {
+pub struct WritePacketState {
-    let packet_type = packet.get_packet_type();
+    pub required_buffer_size: usize,
+    data_frame_length: usize,
+    packet_type: PacketType,
+}
+
+pub(crate) fn start_write_packet(packet: &Packet) -> WritePacketState {
     let data_frame_length = calculate_data_frame_length(&packet);
+    let required_buffer_size = packet_constants::HEADER_SIZE + data_frame_length;
+    let packet_type = get_packet_type(packet);
+    WritePacketState {
+        required_buffer_size,
+        data_frame_length,
+        packet_type,
+    }
+}
 
-    let mut buf = vec![0u8; 3 + data_frame_length];
+pub(crate) fn write_packet(buf: &mut [u8], packet: &Packet, write_state: WritePacketState) -> Result<(), PacketError> {
-    write_header_frame(&mut buf, packet_type, data_frame_length);
+    write_header_frame(buf, write_state.packet_type, write_state.data_frame_length);
-
+    if write_state.data_frame_length > 0 {
-    if data_frame_length > 0 {
         write_data_frame(&mut buf[3..], packet);
     }
 
-    stream.write_all(&buf).map_err(PacketError::TcpStreamError)?;
-    stream.flush().map_err(PacketError::TcpStreamError)?;
-
     Ok(())
 }
 
 fn write_header_frame(header_frame: &mut [u8], packet_type: PacketType, data_frame_length: usize) {
     header_frame[0] = get_byte_from_packet_type(&packet_type);
     if data_frame_length > 0 {
-        write_length(data_frame_length, &mut header_frame[1..3]);
+        write_u16(data_frame_length, &mut header_frame[1..3]);
     }
 }
 
 fn write_data_frame(data_frame: &mut [u8], packet: &Packet) {
     match packet {
         Packet::Connect(connect_packet) => {
-            write_u64(connect_packet.cursor, data_frame);
+            let connect_mode_byte = get_byte_from_connect_mode(&connect_packet.mode);
+            data_frame[0] = connect_mode_byte;
         }
         Packet::Publish(publish_packet) => {
             write_event_data(publish_packet.event_type, &publish_packet.data_utf8, data_frame);
         }
-        Packet::SendEvent(send_event_packet) => {
+        Packet::Deliver(send_event_packet) => {
             write_u64(send_event_packet.event_id, data_frame);
             write_event_data(
                 send_event_packet.event_type,
                 &send_event_packet.data_utf8,
-                &mut data_frame[1..],
+                &mut data_frame[8..],
             );
         }
         Packet::Settle(settle_packet) => {
             write_u64(settle_packet.cursor, data_frame);
         }
-        Packet::Ack(ack_packet) => {
+        Packet::Disconnect => {}
-            data_frame[0] = get_byte_from_packet_type(&ack_packet.packet_type);
-        }
-        _ => {}
     }
 }
 
 fn write_event_data(event_type: u8, data_utf8: &[u8], buf: &mut [u8]) {
     buf[0] = event_type;
-    write_length(data_utf8.len(), &mut buf[1..3]);
+    write_u16(data_utf8.len(), &mut buf[1..3]);
     let available_length = buf.len() - 3;
-    buf[3..].copy_from_slice(&data_utf8[..available_length]);
+    buf[3..].copy_from_slice(&data_utf8[0..available_length]);
 }
 
 fn write_u64(v64: u64, data_frame: &mut [u8]) {
     let bytes = v64.to_le_bytes();
-    data_frame[0..8].copy_from_slice(&bytes);
+    data_frame[0..8].copy_from_slice(&bytes[0..8]);
 }
 
-fn read_u64(stream: &mut TcpStream) -> Result<u64, PacketError> {
+fn write_u16(mut length: usize, buf: &mut [u8]) {
-    let mut le_bytes = [0u8; 8];
-    stream.read_exact(&mut le_bytes).map_err(PacketError::TcpStreamError)?;
-    Ok(u64::from_le_bytes(le_bytes))
-}
-
-fn read_byte(stream: &mut TcpStream) -> Result<u8, PacketError> {
-    let mut byte = [0u8; 1];
-    stream.read_exact(&mut byte).map_err(PacketError::TcpStreamError)?;
-    Ok(byte[0])
-}
-
-fn get_byte_from_packet_type(packet_type: &PacketType) -> u8 {
-    match packet_type {
-        PacketType::Connect => 0,
-        PacketType::Unsubscribe => 1,
-        PacketType::Publish => 2,
-        PacketType::SendEvent => 3,
-        PacketType::Settle => 4,
-        PacketType::Ack => 5,
-        PacketType::Disconnect => 6,
-    }
-}
-
-fn get_packet_type_from_byte(byte: u8) -> Result<PacketType, PacketError> {
-    let packet_type = match byte {
-        0 => Some(PacketType::Connect),
-        1 => Some(PacketType::Unsubscribe),
-        2 => Some(PacketType::Publish),
-        3 => Some(PacketType::SendEvent),
-        4 => Some(PacketType::Settle),
-        5 => Some(PacketType::Ack),
-        6 => Some(PacketType::Disconnect),
-        _ => None,
-    };
-
-    if let Some(packet_type) = packet_type {
-        return Ok(packet_type);
-    }
-
-    Err(PacketError::WrongPacketType(byte))
-}
-
-fn write_length(mut length: usize, buf: &mut [u8]) {
     if length > 0xffff {
         length = 0xffff;
     }
 
     let bytes = length.to_le_bytes();
-    buf[0..2].copy_from_slice(&bytes);
+    buf[0..2].copy_from_slice(&bytes[0..2]);
+}
+
+fn read_u64(buf: &[u8]) -> Result<u64, PacketError> {
+    let mut le_bytes = [0u8; 8];
+    if le_bytes.len() != buf.len() {
+        return Err(PacketError::PacketProtocolError);
+    }
+
+    le_bytes.copy_from_slice(buf);
+    Ok(u64::from_le_bytes(le_bytes))
+}
+
+fn read_u16(buf: &[u8]) -> Result<usize, PacketError> {
+    if buf.len() != 2 {
+        return Err(PacketError::PacketProtocolError);
+    }
+
+    Ok(usize::from_le_bytes([buf[0], buf[1], 0, 0, 0, 0, 0, 0]))
+}
+
+pub fn get_packet_type(packet: &Packet) -> PacketType {
+    match packet {
+        Packet::Connect(_) => PacketType::Connect,
+        Packet::Publish(_) => PacketType::Publish,
+        Packet::Deliver(_) => PacketType::Deliver,
+        Packet::Settle(_) => PacketType::Settle,
+        Packet::Disconnect => PacketType::Disconnect,
+    }
+}
+
+fn get_byte_from_packet_type(packet_type: &PacketType) -> u8 {
+    match packet_type {
+        PacketType::Connect => 0,
+        PacketType::Publish => 1,
+        PacketType::Deliver => 2,
+        PacketType::Settle => 3,
+        PacketType::Disconnect => 4,
+    }
+}
+
+fn get_packet_type_from_byte(byte: u8) -> Result<PacketType, PacketError> {
+    match byte {
+        0 => Ok(PacketType::Connect),
+        1 => Ok(PacketType::Publish),
+        2 => Ok(PacketType::Deliver),
+        3 => Ok(PacketType::Settle),
+        4 => Ok(PacketType::Disconnect),
+        _ => Err(PacketError::WrongPacketType(byte)),
+    }
+}
+
+fn get_byte_from_connect_mode(mode: &ConnectMode) -> u8 {
+    match mode {
+        ConnectMode::Publish => 0,
+        ConnectMode::Subscribe => 1,
+    }
+}
+
+fn get_connect_mode_from_byte(byte: u8) -> Result<ConnectMode, PacketError> {
+    match byte {
+        0 => Ok(ConnectMode::Publish),
+        1 => Ok(ConnectMode::Subscribe),
+        _ => Err(PacketError::PacketProtocolError),
+    }
 }
 
 fn calculate_data_frame_length(packet: &Packet) -> usize {
     let mut ret = match packet {
-        Packet::Connect(_) => 8,
+        Packet::Connect(_) => 1,
-        Packet::Unsubscribe => 0,
         Packet::Publish(publish_packet) => 1 + 2 + publish_packet.data_utf8.len(),
-        Packet::SendEvent(send_event_packet) => 8 + 1 + 2 + send_event_packet.data_utf8.len(),
+        Packet::Deliver(deliver_packet) => 8 + 1 + 2 + deliver_packet.data_utf8.len(),
         Packet::Settle(_) => 8,
-        Packet::Ack(_) => 1,
         Packet::Disconnect => 0,
     };
 

crates/hd-lib/src/error/mod.rs

@@ -1,22 +1,48 @@
-use std::fmt::Display;
+use std::{fmt::Display, string::FromUtf8Error};
+
+use crate::bus::packet::PacketError;
 
 #[derive(Debug)]
-pub enum HError {
+pub enum HdError {
     BusLockPoisoned(String),
     SubscriptionNotFound(String),
-    TcpSocketBindError(std::io::Error),
+    IoError(std::io::Error),
-    TcpPeerError(std::io::Error),
+    TcpPacketError(PacketError),
+    Utf8FormatError(FromUtf8Error),
     NoMoreEvents,
+    ProtocolError,
+    PeerDisconnect,
 }
 
-impl Display for HError {
+impl Display for HdError {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         match self {
-            HError::BusLockPoisoned(s) => write!(f, "{}", s),
+            HdError::BusLockPoisoned(s) => write!(f, "{}", s),
-            HError::SubscriptionNotFound(s) => write!(f, "{}", s),
+            HdError::SubscriptionNotFound(s) => write!(f, "{}", s),
-            HError::TcpSocketBindError(error) => write!(f, "{}", error),
+            HdError::IoError(error) => write!(f, "{}", error),
-            HError::TcpPeerError(error) => write!(f, "{}", error),
+            HdError::NoMoreEvents => write!(f, "{}", "No more events in the bus for now"),
-            HError::NoMoreEvents => write!(f, "{}", "No more events in the bus for now"),
+            HdError::TcpPacketError(packet_error) => write!(f, "Failed to read/write packets due to {}", packet_error),
+            HdError::PeerDisconnect => write!(f, "Peer disconnect"),
+            HdError::Utf8FormatError(e) => write!(f, "Utf8 formatting error {}", e),
+            HdError::ProtocolError => write!(f, "Protocol error"),
         }
     }
 }
+
+impl From<std::io::Error> for HdError {
+    fn from(value: std::io::Error) -> Self {
+        HdError::IoError(value)
+    }
+}
+
+impl From<PacketError> for HdError {
+    fn from(value: PacketError) -> Self {
+        HdError::TcpPacketError(value)
+    }
+}
+
+impl From<FromUtf8Error> for HdError {
+    fn from(value: FromUtf8Error) -> Self {
+        HdError::Utf8FormatError(value)
+    }
+}

crates/hd-lib/src/lib.rs

@@ -1,3 +1,39 @@
+pub mod bus;
 pub mod error;
 pub mod thread_pool;
-pub mod bus;
+
+pub mod hd_tcp {
+    use std::{
+        io::{Read, Write},
+        net::TcpStream,
+    };
+
+    use crate::{
+        bus::packet::{self, Packet},
+        error::HdError,
+    };
+
+    pub fn read_packet(stream: &mut TcpStream) -> Result<Packet, HdError> {
+        let mut header_buf = [0u8; 3];
+        stream.read_exact(&mut header_buf)?;
+        let read_state = packet::start_read_packet(&mut header_buf)?;
+
+        let mut buf = vec![0u8; read_state.required_buffer_size];
+        if read_state.required_buffer_size > 0 {
+            stream.read_exact(&mut buf)?;
+        }
+
+        Ok(packet::read_packet(&buf[0..], read_state)?)
+    }
+
+    pub fn write_packet(stream: &mut TcpStream, packet: &Packet) -> Result<(), HdError> {
+        let write_state = packet::start_write_packet(packet);
+        let mut buf = vec![0u8; write_state.required_buffer_size];
+        packet::write_packet(&mut buf, packet, write_state)?;
+
+        stream.write_all(&mut buf)?;
+        stream.flush()?;
+
+        Ok(())
+    }
+}

crates/hd-lib/src/thread_pool/mod.rs

@@ -6,9 +6,9 @@ use std::{
     thread::{self, JoinHandle},
 };
 
-use tracing::{Level, info, span};
+use tracing::{Level, error, info, span};
 
-use crate::error::HError;
+use crate::error::HdError;
 
 pub struct ThreadPool {
     workers: Vec<Worker>,
@@ -21,7 +21,7 @@ struct Worker {
 }
 
 struct Job {
-    f: Box<dyn FnOnce() -> Result<(), HError> + Send>,
+    f: Box<dyn FnOnce() -> Result<(), HdError> + Send>,
 }
 
 impl ThreadPool {
@@ -36,9 +36,9 @@ impl ThreadPool {
         Self { workers, tx }
     }
 
-    pub fn execute<F>(&self, f: F) -> Result<(), HError>
+    pub fn execute<F>(&self, f: F) -> Result<(), HdError>
     where
-        F: FnOnce() -> Result<(), HError>,
+        F: FnOnce() -> Result<(), HdError>,
         F: Send + 'static,
     {
         let job = Job { f: Box::new(f) };
@@ -54,8 +54,12 @@ impl ThreadPool {
             loop {
                 let job = rx.lock().unwrap().recv().unwrap(); // Lock is released here
                 info!("Received a new job, executing");
-                (job.f)().unwrap(); // TODO: Panic recovery
+                let result = (job.f)();
-                info!("Finished executing job");
+                if let Err(e) = result {
+                    error!("Job failed to complete successfully due to error {}", e);
+                }
+
+                info!("Job execution complete, worker thread is free");
             }
         });
 

crates/hd-server/Cargo.toml

@@ -5,5 +5,4 @@ edition = "2024"
 
 [dependencies]
 tracing = { workspace = true }
-tracing-subscriber = { version = "0.3", features = ["env-filter", "fmt"] }
+hd-lib = {path = "../hd-lib"}
-hd-lib = { path = "../hd-lib" }

crates/hd-server/src/connection.rs

@@ -0,0 +1,185 @@
+use std::{net::TcpStream, time::Duration};
+
+use hd_lib::{
+    bus::{packet::*, *},
+    error::HdError,
+    hd_tcp,
+    thread_pool::ThreadPool,
+};
+use tracing::{debug, error, info, info_span};
+
+#[derive(Copy, Clone, PartialEq)]
+enum HdClientConnectionState {
+    Connect,
+    Deliver,
+    Wait,
+    Disconnect,
+}
+
+struct HdClientConnection {
+    pub addr: String,
+    stream: TcpStream,
+    state: HdClientConnectionState,
+    subscription: Option<SubscriptionInfo>,
+    bus: EventBus,
+}
+
+impl std::fmt::Display for HdClientConnectionState {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        let to_write = match self {
+            HdClientConnectionState::Connect => "CONNECT",
+            HdClientConnectionState::Deliver => "DELIVER",
+            HdClientConnectionState::Wait => "WAIT",
+            HdClientConnectionState::Disconnect => "DISCONNECT",
+        };
+
+        write!(f, "{}", to_write)
+    }
+}
+
+impl HdClientConnection {
+    fn new(bus: EventBus, stream: TcpStream) -> Result<Self, HdError> {
+        let addr = stream
+            .peer_addr()
+            .map(|a| a.to_string())
+            .unwrap_or("unknown".to_string());
+
+        stream.set_read_timeout(Some(Duration::from_secs(5)))?;
+
+        let state = HdClientConnectionState::Connect;
+
+        Ok(Self {
+            stream,
+            addr,
+            state,
+            bus,
+            subscription: None,
+        })
+    }
+}
+
+pub(crate) fn handle_client_connection(
+    thread_pool: &ThreadPool,
+    bus: EventBus,
+    stream: TcpStream,
+) -> Result<(), HdError> {
+    thread_pool.execute(move || {
+        let mut client = HdClientConnection::new(bus, stream)?;
+
+        let _scope = info_span!("handle_client", addr = client.addr,).entered();
+
+        loop {
+            if let HdClientConnectionState::Disconnect = client.state {
+                unsubscribe_client(&mut client)?;
+                info!("Closing client connection");
+                return Ok(());
+            }
+
+            let prev_state = client.state;
+            let next_state = match handle_event_bus_client_state(&mut client) {
+                Ok(s) => s,
+                Err(e) => {
+                    error!("Error while handling client {}", e);
+                    HdClientConnectionState::Disconnect
+                }
+            };
+
+            if prev_state != next_state {
+                info!("State change {} -> {}", prev_state, next_state);
+            }
+
+            client.state = next_state;
+        }
+    })
+}
+
+fn handle_event_bus_client_state(client: &mut HdClientConnection) -> Result<HdClientConnectionState, HdError> {
+    match client.state {
+        HdClientConnectionState::Connect => handle_state_connect(client),
+        HdClientConnectionState::Deliver => handle_state_deliver_event(client),
+        HdClientConnectionState::Wait => handle_state_wait_event(client),
+        HdClientConnectionState::Disconnect => Ok(HdClientConnectionState::Disconnect),
+    }
+}
+
+fn handle_state_connect(client: &mut HdClientConnection) -> Result<HdClientConnectionState, HdError> {
+    match hd_tcp::read_packet(&mut client.stream)? {
+        Packet::Connect(connect_packet) => {
+            info!("Received connect");
+            let next = match connect_packet.mode {
+                ConnectMode::Publish => HdClientConnectionState::Wait,
+                ConnectMode::Subscribe => {
+                    let subscription = client.bus.subscribe()?;
+                    client.subscription = Some(subscription);
+                    HdClientConnectionState::Deliver
+                }
+            };
+
+            Ok(next)
+        }
+        p => {
+            error!(
+                "Received wrong packet of type {}, expected {}. Will disconnect",
+                get_packet_type(&p),
+                PacketType::Connect
+            );
+            Ok(HdClientConnectionState::Disconnect)
+        }
+    }
+}
+
+fn handle_state_deliver_event(client: &mut HdClientConnection) -> Result<HdClientConnectionState, HdError> {
+    if let Some(subscription) = &mut client.subscription {
+        // TODO: When client sends unsubscribe, server shouldn't take forever
+        // to ack that.
+        let Ok(evt) = subscription.event_recv.recv() else {
+            error!("Failed to receive a message for this subscription, disconnecting");
+            return Ok(HdClientConnectionState::Disconnect);
+        };
+
+        let packet = Packet::create_deliver_packet(evt);
+        hd_tcp::write_packet(&mut client.stream, &packet)?;
+        debug!("Published event to client, waiting for ack");
+
+        let response_packet = hd_tcp::read_packet(&mut client.stream)?;
+        if let Packet::Disconnect = &response_packet {
+            info!("Client disconnected");
+            // TODO: Allow unsubscribe, and actually disconnect
+            return Ok(HdClientConnectionState::Disconnect);
+        }
+
+        debug!("Publish ack received, message is settled");
+    }
+
+    Ok(HdClientConnectionState::Deliver)
+}
+
+fn unsubscribe_client(client: &mut HdClientConnection) -> Result<(), HdError> {
+    if let Some(subscription) = &client.subscription {
+        client.bus.unsubscribe(subscription.id)?;
+        client.subscription = None;
+    };
+
+    Ok(())
+}
+
+fn handle_state_wait_event(client: &mut HdClientConnection) -> Result<HdClientConnectionState, HdError> {
+    match hd_tcp::read_packet(&mut client.stream)? {
+        Packet::Publish(publish_packet) => {
+            debug!("Received event from client, publishing to bus");
+            let data = String::from_utf8(publish_packet.data_utf8)?;
+            let evt = EventData::new(publish_packet.event_type, &data);
+            client.bus.publish(evt)?;
+
+            Ok(HdClientConnectionState::Wait)
+        }
+        p => {
+            error!(
+                "Received wrong packet of type {}, expected {}. Will disconnect",
+                get_packet_type(&p),
+                PacketType::Connect
+            );
+            Ok(HdClientConnectionState::Disconnect)
+        }
+    }
+}

crates/hd-server/src/lib.rs

@@ -1,24 +1,21 @@
-use std::{
+mod connection;
-    io::{Read, Write},
-    net::TcpStream,
-    sync::{Arc, mpsc::Receiver},
-    time::Duration,
-};
 
 use hd_lib::{
-    bus::{Event, EventBus, EventData}, error::HError, thread_pool::ThreadPool,
+    bus::{EventBus, EventData},
+    error::HdError,
+    thread_pool::ThreadPool,
 };
 
-use tracing::{Span, info, info_span};
+use tracing::{error, info};
 
 use std::net::TcpListener;
 
-pub struct TcpEventBus {
+pub struct HdBus {
     bus: EventBus,
     pool: ThreadPool,
 }
 
-impl TcpEventBus {
+impl HdBus {
     pub fn new() -> Self {
         let pool = ThreadPool::new(10);
         Self {
@@ -27,182 +24,22 @@ impl TcpEventBus {
         }
     }
 
-    pub fn publish(&self, evt: EventData) -> Result<(), HError> {
+    pub fn publish(&self, evt: EventData) -> Result<(), HdError> {
         self.bus.publish(evt)
     }
 }
 
-impl TcpEventBus {
+impl HdBus {
-    pub fn start(&self, addr: &'static str) -> Result<(), HError> {
+    pub fn start(&self, addr: &'static str) -> Result<(), HdError> {
-        let listener = TcpListener::bind(addr).map_err(HError::TcpSocketBindError)?;
+        let listener = TcpListener::bind(addr)?;
-        info!("Heimdall listening at {}", addr);
+        info!("Heimdall running at {}", addr);
         for stream in listener.incoming() {
             match stream {
-                Ok(stream) => handle_event_bus_client(&self.pool, self.bus.clone(), stream)?,
+                Ok(stream) => connection::handle_client_connection(&self.pool, self.bus.clone(), stream)?,
-                Err(e) => tracing::error!("TCP connection failed {}", e),
+                Err(e) => error!("TCP connection failed {}", e),
             }
         }
 
         Ok(())
     }
 }
-
-#[derive(Copy, Clone, PartialEq)]
-enum EventBusClientConnectionState {
-    Connect,
-    ReadPrelude,
-    ReadCursor,
-    SendEvent,
-    ReceiveAck,
-    Disconnect,
-}
-
-struct EventBusClientConnection {
-    stream: TcpStream,
-    pub addr: String,
-    state: EventBusClientConnectionState,
-    subscription: Option<Receiver<Arc<Event>>>,
-    bus: EventBus,
-}
-
-impl EventBusClientConnection {
-    fn new(bus: EventBus, stream: TcpStream) -> Result<Self, HError> {
-        let addr = stream
-            .peer_addr()
-            .map(|a| a.to_string())
-            .unwrap_or("unknown".to_string());
-
-        stream
-            .set_read_timeout(Some(Duration::from_secs(5)))
-            .map_err(HError::TcpPeerError)?;
-
-        let state = EventBusClientConnectionState::Connect;
-
-        Ok(Self {
-            stream,
-            addr,
-            state,
-            bus,
-            subscription: None,
-        })
-    }
-}
-
-pub fn handle_event_bus_client(thread_pool: &ThreadPool, bus: EventBus, stream: TcpStream) -> Result<(), HError> {
-    thread_pool.execute(move || {
-        let mut client = EventBusClientConnection::new(bus, stream)?;
-
-        let _scope = info_span!(
-            "handle_event_bus_client",
-            addr = client.addr,
-            state = %client.state
-        )
-        .entered();
-
-        loop {
-            if let EventBusClientConnectionState::Disconnect = client.state {
-                info!("Closing connection due to invalid input");
-                return Ok(());
-            }
-
-            let prev_state = client.state;
-            handle_event_bus_client_state(&mut client)
-                .inspect_err(|e| info!("Closing connection due to error {}", e))?;
-            if prev_state != client.state {
-                info!("State change {} -> {}", prev_state, client.state);
-            }
-
-            Span::current().record("state", format!("{}", client.state));
-        }
-    })
-}
-
-fn handle_event_bus_client_state(client: &mut EventBusClientConnection) -> Result<(), HError> {
-    let next_state = match client.state {
-        EventBusClientConnectionState::Connect => handle_state_connect(client),
-        EventBusClientConnectionState::ReadPrelude => todo!(),
-        EventBusClientConnectionState::ReadCursor => todo!(),
-        EventBusClientConnectionState::SendEvent => handle_state_send_event(client),
-        EventBusClientConnectionState::ReceiveAck => todo!(),
-        _ => Ok(EventBusClientConnectionState::Disconnect),
-    }?;
-
-    client.state = next_state;
-    Ok(())
-}
-
-fn handle_state_connect(client: &mut EventBusClientConnection) -> Result<EventBusClientConnectionState, HError> {
-    let mut buf = [0u8; 1];
-    client.stream.read_exact(&mut buf).map_err(HError::TcpPeerError)?;
-    if buf[0] == 1 {
-        let subscription = client.bus.subscribe()?;
-        client.subscription = Some(subscription);
-        return Ok(EventBusClientConnectionState::SendEvent);
-    }
-
-    Ok(EventBusClientConnectionState::Disconnect)
-}
-
-fn handle_state_send_event(client: &mut EventBusClientConnection) -> Result<EventBusClientConnectionState, HError> {
-    if let Some(subscription) = &mut client.subscription {
-        loop {
-            let Ok(evt) = subscription.recv() else {
-                return Ok(EventBusClientConnectionState::Disconnect);
-            };
-
-            let data_len = evt.data.data_utf8.len();
-            // [ ID  ] [TYPE] [ LEN ] [ DATA ]
-            // 0  to 7   8    9 to 15 . . . .
-            let buf_length = data_len + 17;
-            let mut buf = vec![0u8; buf_length];
-            write_le_bytes(evt.id, &mut buf[0..8]);
-            write_le_bytes(data_len as u64, &mut buf[9..17]);
-            buf[8] = evt.data.event_type;
-
-            for i in 17..buf_length {
-                buf[i] = evt.data.data_utf8[i - 17];
-            }
-
-            info!(
-                "SEND => ID: {} TYPE: {} LEN: {} <=",
-                evt.id, evt.data.event_type, data_len
-            );
-
-            let stream = &mut client.stream;
-            stream.write(&mut buf).map_err(HError::TcpPeerError)?;
-            stream.flush().map_err(HError::TcpPeerError)?;
-
-            let mut ack = vec![0];
-            stream.read(&mut ack).map_err(HError::TcpPeerError)?;
-        }
-    }
-
-    Ok(EventBusClientConnectionState::Disconnect)
-}
-
-fn write_le_bytes(v64: u64, buf: &mut [u8]) {
-    let le_bytes = v64.to_le_bytes();
-    buf[0] = le_bytes[0];
-    buf[1] = le_bytes[1];
-    buf[2] = le_bytes[2];
-    buf[3] = le_bytes[3];
-    buf[4] = le_bytes[4];
-    buf[5] = le_bytes[5];
-    buf[6] = le_bytes[6];
-    buf[7] = le_bytes[7];
-}
-
-impl std::fmt::Display for EventBusClientConnectionState {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        let to_write = match self {
-            EventBusClientConnectionState::Connect => "CONNECT",
-            EventBusClientConnectionState::ReadPrelude => "READ_PRELUDE",
-            EventBusClientConnectionState::ReadCursor => "READ_CURSOR",
-            EventBusClientConnectionState::SendEvent => "SEND_EVENT",
-            EventBusClientConnectionState::ReceiveAck => "RECEIVE_ACK",
-            EventBusClientConnectionState::Disconnect => "DISCONNECT",
-        };
-
-        write!(f, "{}", to_write)
-    }
-}