Refactors

crates/hd-client/Cargo.toml

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

crates/hd-client/src/lib.rs

@@ -1,43 +1,42 @@
+
 use std::net::TcpStream;
 
 use hd_lib::{
-    bus::packet::{self, Packet, PacketType},
-    error::HError,
+    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(HError),
+    HError(HdError),
 }
 
-impl From<HError> for EventBusClientError {
-    fn from(value: HError) -> Self {
+impl From<HdError> for HdClientError {
+    fn from(value: HdError) -> Self {
         Self::HError(value)
     }
 }
 
-pub struct EventBusClient {
+pub struct HdClient {
     tcp_stream: TcpStream,
 }
 
-impl EventBusClient {
-    pub fn start(addr: &'static str) -> Result<Self, EventBusClientError> {
-        let mut stream = TcpStream::connect(addr).map_err(EventBusClientError::TcpStreamError)?;
-        hd_tcp::write_packet(&mut stream, &Packet::create_connect_packet())?;
-        hd_tcp::read_ack(&mut stream, &[PacketType::Connect])?;
-
+impl HdClient {
+    pub fn start(addr: &'static str) -> Result<Self, HdClientError> {
+        let stream = TcpStream::connect(addr).map_err(HdClientError::TcpStreamError)?;
         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> {
         debug!("Going to subscribe");
-        hd_tcp::write_packet(&mut self.tcp_stream, &Packet::create_subscribe_packet(0))?;
-        hd_tcp::read_ack(&mut self.tcp_stream, &[PacketType::Subscribe])?;
-
+        hd_tcp::write_packet(
+            &mut self.tcp_stream,
+            &Packet::create_connect_packet(packet::ConnectMode::Subscribe),
+        )?;
         loop {
             let packet = hd_tcp::read_packet(&mut self.tcp_stream)?;
             if let Packet::Disconnect = packet {
@@ -46,7 +45,7 @@ impl EventBusClient {
             }
 
             debug!("Received packet ... {}", packet::get_packet_type(&packet));
-            hd_tcp::write_ack(&mut self.tcp_stream, PacketType::SendEvent)?;
+            hd_tcp::write_packet(&mut self.tcp_stream, &Packet::create_settle_packet(1))?;
         }
     }
 }

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

@@ -1,11 +1,11 @@
-use hd_client::{EventBusClient, EventBusClientError};
+use hd_client::{HdClient, HdClientError};
 
-fn main() -> Result<(), EventBusClientError> {
+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 = EventBusClient::start(addr)?;
+    let mut client = HdClient::start(addr)?;
     client.subscribe()
 }

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

@@ -1,16 +1,16 @@
 use std::{sync::Arc, thread, time::Duration};
 
-use hd_lib::{bus::EventData, error::HError};
-use hd_server::TcpEventBus;
+use hd_lib::{bus::EventData, error::HdError};
+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 server = Arc::new(TcpEventBus::new());
+    let server = Arc::new(HdBus::new());
     server.publish(EventData::new(1, "Hello"))?;
     server.publish(EventData::new(1, "World"))?;
 

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

@@ -1,8 +1,8 @@
 pub mod packet;
 
-use tracing::{debug_span, 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> {
-        let mut bus = self.inner.write().map_err(|e| HError::BusLockPoisoned(e.to_string()))?;
+    pub fn publish(&self, evt: EventData) -> Result<(), HdError> {
+        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 {
@@ -103,8 +108,8 @@ impl EventBus {
         Ok(())
     }
 
-    pub fn get_next_event(&self, cursor: Cursor) -> Result<Arc<Event>, HError> {
-        let bus = self.inner.read().map_err(|e| HError::BusLockPoisoned(e.to_string()))?;
+    pub fn get_next_event(&self, cursor: Cursor) -> Result<Arc<Event>, HdError> {
+        let bus = self.inner.read().map_err(|e| HdError::BusLockPoisoned(e.to_string()))?;
 
         for evt in &bus.events {
             if evt.id >= cursor {
@@ -112,22 +117,22 @@ impl EventBus {
             }
         }
 
-        Err(HError::NoMoreEvents)
+        Err(HdError::NoMoreEvents)
     }
 
-    pub fn subscribe(&self) -> Result<Receiver<Arc<Event>>, HError> {
-        let mut bus = self.inner.write().map_err(|e| HError::BusLockPoisoned(e.to_string()))?;
+    pub fn subscribe(&self) -> Result<SubscriptionInfo, HdError> {
+        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,
@@ -137,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> {
-        let mut bus = self.inner.write().map_err(|e| HError::BusLockPoisoned(e.to_string()))?;
+    pub fn unsubscribe(&self, id: SubscriptionId) -> Result<(), HdError> {
+        let mut bus = self.inner.write().map_err(|e| HdError::BusLockPoisoned(e.to_string()))?;
 
         let mut found = None;
         let mut index = 0;
@@ -155,7 +166,7 @@ impl EventBus {
         }
 
         match found {
-            None => Err(HError::SubscriptionNotFound(format!(
+            None => Err(HdError::SubscriptionNotFound(format!(
                 "Subscription with id {} is not found",
                 id
             ))),
@@ -193,8 +204,9 @@ 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 = debug_span!("subscription_delivery", subscription.id).entered();
             loop {

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

@@ -9,25 +9,25 @@ mod packet_constants {
 #[derive(PartialEq)]
 pub enum PacketType {
     Connect,
-    Subscribe,
-    Unsubscribe,
     Publish,
-    SendEvent,
+    Deliver,
     Settle,
-    Ack,
     Disconnect,
 }
 
+#[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::Subscribe => "SUBSCRIBE",
-            PacketType::Unsubscribe => "UNSUBSCRIBE",
             PacketType::Publish => "PUBLISH",
-            PacketType::SendEvent => "SEND_EVENT",
-            PacketType::Settle => "SETTLE_EVENT",
-            PacketType::Ack => "ACK",
+            PacketType::Deliver => "DELIVER",
+            PacketType::Settle => "SETTLE",
             PacketType::Disconnect => "DISCONNECT",
         };
 
@@ -36,18 +36,15 @@ impl Display for PacketType {
 }
 
 pub enum Packet {
-    Connect,
-    Subscribe(SubscribePacket),
-    Unsubscribe,
+    Connect(ConnectPacket),
     Publish(PublishPacket),
-    SendEvent(SendEventPacket),
+    Deliver(DeliverPacket),
     Settle(SettlePacket),
-    Ack(AckPacket),
     Disconnect,
 }
 
-pub struct SubscribePacket {
-    cursor: u64,
+pub struct ConnectPacket {
+    pub mode: ConnectMode,
 }
 
 pub struct PublishPacket {
@@ -55,7 +52,7 @@ pub struct PublishPacket {
     pub data_utf8: Vec<u8>,
 }
 
-pub struct SendEventPacket {
+pub struct DeliverPacket {
     event_id: u64,
     event_type: u8,
     data_utf8: Vec<u8>,
@@ -70,16 +67,8 @@ pub struct AckPacket {
 }
 
 impl Packet {
-    pub fn create_connect_packet() -> Packet {
-        Packet::Connect
-    }
-
-    pub fn create_subscribe_packet(cursor: u64) -> Packet {
-        Packet::Subscribe(SubscribePacket { cursor })
-    }
-
-    pub fn create_unsubscribe_packet() -> Packet {
-        Packet::Unsubscribe
+    pub fn create_connect_packet(mode: ConnectMode) -> Packet {
+        Packet::Connect(ConnectPacket { mode })
     }
 
     // TODO: Avoid copying data_utf8
@@ -91,8 +80,8 @@ impl Packet {
     }
 
     // TODO: Avoid copying data_utf8
-    pub fn create_send_event_packet(evt: Arc<Event>) -> Packet {
-        Packet::SendEvent(SendEventPacket {
+    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(),
@@ -103,10 +92,6 @@ impl Packet {
         Packet::Settle(SettlePacket { cursor })
     }
 
-    pub fn create_ack_packet(packet_type: PacketType) -> Packet {
-        Packet::Ack(AckPacket { packet_type })
-    }
-
     pub fn create_disconnect_packet() -> Packet {
         Packet::Disconnect
     }
@@ -134,7 +119,7 @@ pub struct ReadPacketState {
     packet_type: PacketType,
 }
 
-pub fn start_read_packet(buf: &mut [u8; packet_constants::HEADER_SIZE]) -> Result<ReadPacketState, PacketError> {
+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])?;
 
@@ -144,14 +129,13 @@ pub fn start_read_packet(buf: &mut [u8; packet_constants::HEADER_SIZE]) -> Resul
     })
 }
 
-pub fn read_packet(buf: &[u8], read_state: ReadPacketState) -> Result<Packet, PacketError> {
+pub(crate) fn read_packet(buf: &[u8], read_state: ReadPacketState) -> Result<Packet, PacketError> {
     match read_state.packet_type {
-        PacketType::Connect => Ok(Packet::Connect),
-        PacketType::Subscribe => {
-            let cursor = read_u64(&buf[0..])?;
-            Ok(Packet::Subscribe(SubscribePacket { cursor }))
+        PacketType::Connect => {
+            let mode_byte = buf[0];
+            let mode = get_connect_mode_from_byte(mode_byte)?;
+            Ok(Packet::create_connect_packet(mode))
         }
-        PacketType::Unsubscribe => Ok(Packet::Unsubscribe),
         PacketType::Publish => {
             let event_type = buf[0];
             let event_len = read_u16(&buf[1..3])?;
@@ -163,27 +147,23 @@ pub fn read_packet(buf: &[u8], read_state: ReadPacketState) -> Result<Packet, Pa
             };
             Ok(Packet::Publish(publish_packet))
         }
-        PacketType::SendEvent => {
+        PacketType::Deliver => {
             let event_id = read_u64(&buf[0..8])?;
             let event_type = buf[8];
             let event_len = read_u16(&buf[9..11])?;
 
-            let send_event_packet = SendEventPacket {
+            let deliver_packet = DeliverPacket {
                 event_id,
                 event_type,
                 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(&buf[0..8])?;
-            Ok(Packet::Settle(SettlePacket { cursor }))
+            Ok(Packet::create_settle_packet(cursor))
         }
-        PacketType::Ack => {
-            let packet_type = get_packet_type_from_byte(buf[0])?;
-            Ok(Packet::Ack(AckPacket { packet_type }))
-        }
-        PacketType::Disconnect => Ok(Packet::Disconnect),
+        PacketType::Disconnect => Ok(Packet::create_disconnect_packet()),
     }
 }
 
@@ -193,7 +173,7 @@ pub struct WritePacketState {
     packet_type: PacketType,
 }
 
-pub fn start_write_packet(packet: &Packet) -> WritePacketState {
+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);
@@ -204,7 +184,7 @@ pub fn start_write_packet(packet: &Packet) -> WritePacketState {
     }
 }
 
-pub fn write_packet(buf: &mut [u8], packet: &Packet, write_state: WritePacketState) -> Result<(), PacketError> {
+pub(crate) fn write_packet(buf: &mut [u8], packet: &Packet, write_state: WritePacketState) -> Result<(), PacketError> {
     write_header_frame(buf, write_state.packet_type, write_state.data_frame_length);
     if write_state.data_frame_length > 0 {
         write_data_frame(&mut buf[3..], packet);
@@ -222,13 +202,14 @@ fn write_header_frame(header_frame: &mut [u8], packet_type: PacketType, data_fra
 
 fn write_data_frame(data_frame: &mut [u8], packet: &Packet) {
     match packet {
-        Packet::Subscribe(connect_packet) => {
-            write_u64(connect_packet.cursor, data_frame);
+        Packet::Connect(connect_packet) => {
+            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,
@@ -239,10 +220,7 @@ fn write_data_frame(data_frame: &mut [u8], packet: &Packet) {
         Packet::Settle(settle_packet) => {
             write_u64(settle_packet.cursor, data_frame);
         }
-        Packet::Ack(ack_packet) => {
-            data_frame[0] = get_byte_from_packet_type(&ack_packet.packet_type);
-        }
-        _ => {}
+        Packet::Disconnect => {}
     }
 }
 
@@ -287,13 +265,10 @@ fn read_u16(buf: &[u8]) -> Result<usize, PacketError> {
 
 pub fn get_packet_type(packet: &Packet) -> PacketType {
     match packet {
-        Packet::Connect => PacketType::Connect,
-        Packet::Subscribe(_) => PacketType::Subscribe,
-        Packet::Unsubscribe => PacketType::Unsubscribe,
+        Packet::Connect(_) => PacketType::Connect,
         Packet::Publish(_) => PacketType::Publish,
-        Packet::SendEvent(_) => PacketType::SendEvent,
+        Packet::Deliver(_) => PacketType::Deliver,
         Packet::Settle(_) => PacketType::Settle,
-        Packet::Ack(_) => PacketType::Ack,
         Packet::Disconnect => PacketType::Disconnect,
     }
 }
@@ -301,45 +276,45 @@ pub fn get_packet_type(packet: &Packet) -> PacketType {
 fn get_byte_from_packet_type(packet_type: &PacketType) -> u8 {
     match packet_type {
         PacketType::Connect => 0,
-        PacketType::Subscribe => 1,
-        PacketType::Unsubscribe => 2,
-        PacketType::Publish => 3,
-        PacketType::SendEvent => 4,
-        PacketType::Settle => 5,
-        PacketType::Ack => 6,
-        PacketType::Disconnect => 7,
+        PacketType::Publish => 1,
+        PacketType::Deliver => 2,
+        PacketType::Settle => 3,
+        PacketType::Disconnect => 4,
     }
 }
 
 fn get_packet_type_from_byte(byte: u8) -> Result<PacketType, PacketError> {
-    let packet_type = match byte {
-        0 => Some(PacketType::Connect),
-        1 => Some(PacketType::Subscribe),
-        2 => Some(PacketType::Unsubscribe),
-        3 => Some(PacketType::Publish),
-        4 => Some(PacketType::SendEvent),
-        5 => Some(PacketType::Settle),
-        6 => Some(PacketType::Ack),
-        7 => Some(PacketType::Disconnect),
-        _ => None,
-    };
-
-    if let Some(packet_type) = packet_type {
-        return Ok(packet_type);
+    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)),
     }
+}
 
-    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 => 0,
-        Packet::Subscribe(_) => 8,
-        Packet::Unsubscribe => 0,
+        Packet::Connect(_) => 1,
         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

@@ -3,7 +3,7 @@ use std::{fmt::Display, string::FromUtf8Error};
 use crate::bus::packet::PacketError;
 
 #[derive(Debug)]
-pub enum HError {
+pub enum HdError {
     BusLockPoisoned(String),
     SubscriptionNotFound(String),
     IoError(std::io::Error),
@@ -14,35 +14,35 @@ pub enum HError {
     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),
-            HError::SubscriptionNotFound(s) => write!(f, "{}", s),
-            HError::IoError(error) => write!(f, "{}", error),
-            HError::NoMoreEvents => write!(f, "{}", "No more events in the bus for now"),
-            HError::TcpPacketError(packet_error) => write!(f, "Failed to read/write packets due to {}", packet_error),
-            HError::PeerDisconnect => write!(f, "Peer disconnect"),
-            HError::Utf8FormatError(e) => write!(f, "Utf8 formatting error {}", e),
-            HError::ProtocolError => write!(f, "Protocol error"),
+            HdError::BusLockPoisoned(s) => write!(f, "{}", s),
+            HdError::SubscriptionNotFound(s) => write!(f, "{}", s),
+            HdError::IoError(error) => write!(f, "{}", error),
+            HdError::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 HError {
+impl From<std::io::Error> for HdError {
     fn from(value: std::io::Error) -> Self {
-        HError::IoError(value)
+        HdError::IoError(value)
     }
 }
 
-impl From<PacketError> for HError {
+impl From<PacketError> for HdError {
     fn from(value: PacketError) -> Self {
-        HError::TcpPacketError(value)
+        HdError::TcpPacketError(value)
     }
 }
 
-impl From<FromUtf8Error> for HError {
+impl From<FromUtf8Error> for HdError {
     fn from(value: FromUtf8Error) -> Self {
-        HError::Utf8FormatError(value)
+        HdError::Utf8FormatError(value)
     }
 }

crates/hd-lib/src/lib.rs

@@ -9,11 +9,11 @@ pub mod hd_tcp {
     };
 
     use crate::{
-        bus::packet::{self, Packet, PacketType},
-        error::HError,
+        bus::packet::{self, Packet},
+        error::HdError,
     };
 
-    pub fn read_packet(stream: &mut TcpStream) -> Result<Packet, HError> {
+    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)?;
@@ -22,34 +22,11 @@ pub mod hd_tcp {
         if read_state.required_buffer_size > 0 {
             stream.read_exact(&mut buf)?;
         }
-        let packet = packet::read_packet(&buf[0..], read_state)?;
 
-        if let Packet::Disconnect = packet {
-            return Err(HError::PeerDisconnect);
-        };
-
-        Ok(packet)
+        Ok(packet::read_packet(&buf[0..], read_state)?)
     }
 
-    pub fn read_ack(stream: &mut TcpStream, packet_types: &[PacketType]) -> Result<Packet, HError> {
-        let packet = read_packet(stream)?;
-        if let Packet::Ack(ack_packet) = &packet {
-            for packet_type in packet_types.iter() {
-                if *packet_type == ack_packet.packet_type {
-                    return Ok(packet);
-                }
-            }
-        }
-
-        Err(HError::ProtocolError)
-    }
-
-    pub fn write_ack(stream: &mut TcpStream, packet_type: PacketType) -> Result<(), HError> {
-        let packet = Packet::create_ack_packet(packet_type);
-        write_packet(stream, &packet)
-    }
-
-    pub fn write_packet(stream: &mut TcpStream, packet: &Packet) -> Result<(), HError> {
+    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)?;

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
-                info!("Finished executing job");
+                let result = (job.f)();
+                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,29 +1,21 @@
-use std::{
-    net::TcpStream,
-    sync::{Arc, mpsc::Receiver},
-    time::Duration,
-};
+mod connection;
 
 use hd_lib::{
-    bus::{
-        Event, EventBus, EventData,
-        packet::{self, Packet, PacketType, PublishPacket, get_packet_type},
-    },
-    error::HError,
-    hd_tcp,
+    bus::{EventBus, EventData},
+    error::HdError,
     thread_pool::ThreadPool,
 };
 
-use tracing::{debug, error, 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 {
@@ -32,18 +24,18 @@ impl TcpEventBus {
         }
     }
 
-    pub fn publish(&self, evt: EventData) -> Result<(), HError> {
+    pub fn publish(&self, evt: EventData) -> Result<(), HdError> {
         self.bus.publish(evt)
     }
 }
 
-impl TcpEventBus {
-    pub fn start(&self, addr: &'static str) -> Result<(), HError> {
+impl HdBus {
+    pub fn start(&self, addr: &'static str) -> Result<(), HdError> {
         let listener = TcpListener::bind(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) => error!("TCP connection failed {}", e),
             }
         }
@@ -51,199 +43,3 @@ impl TcpEventBus {
         Ok(())
     }
 }
-
-#[derive(Copy, Clone, PartialEq)]
-enum EventBusClientConnectionState {
-    Connect,
-    Command,
-    SendEvent,
-    WaitEvent,
-    Disconnect,
-}
-
-struct EventBusClientConnection {
-    pub addr: String,
-    stream: TcpStream,
-    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)))?;
-
-        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_client", addr = client.addr,).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);
-            }
-        }
-    })
-}
-
-fn handle_event_bus_client_state(client: &mut EventBusClientConnection) -> Result<(), HError> {
-    let next_state = match client.state {
-        EventBusClientConnectionState::Connect => handle_state_connect(client),
-        EventBusClientConnectionState::Command => handle_state_command(client),
-        EventBusClientConnectionState::SendEvent => handle_state_send_event(client),
-        EventBusClientConnectionState::WaitEvent => handle_state_wait_event(client),
-        EventBusClientConnectionState::Disconnect => Ok(EventBusClientConnectionState::Disconnect),
-    }?;
-
-    client.state = next_state;
-    Ok(())
-}
-
-fn handle_state_connect(client: &mut EventBusClientConnection) -> Result<EventBusClientConnectionState, HError> {
-    match hd_tcp::read_packet(&mut client.stream)? {
-        Packet::Connect => {
-            info!("Received connect, acking");
-
-            hd_tcp::write_ack(&mut client.stream, PacketType::Connect)?;
-            info!("Connect ack complete");
-
-            Ok(EventBusClientConnectionState::Command)
-        }
-        p => {
-            error!(
-                "Received wrong packet of type {}, expected {}. Will disconnect",
-                get_packet_type(&p),
-                PacketType::Connect
-            );
-            Ok(EventBusClientConnectionState::Disconnect)
-        }
-    }
-}
-
-fn handle_state_command(client: &mut EventBusClientConnection) -> Result<EventBusClientConnectionState, HError> {
-    match hd_tcp::read_packet(&mut client.stream)? {
-        // TODO: Read cursor
-        Packet::Subscribe(_) => {
-            info!("Received subscribe");
-            let subscription = client.bus.subscribe()?;
-
-            info!("Subscribed to the event bus");
-            client.subscription = Some(subscription);
-
-            // TODO: Return error if any
-            hd_tcp::write_ack(&mut client.stream, PacketType::Subscribe)?;
-            info!("Subscribe ack complete");
-            Ok(EventBusClientConnectionState::SendEvent)
-        }
-        Packet::Publish(publish_packet) => {
-            info!("Received publish packet");
-            publish_packet_to_bus(client, publish_packet)?;
-
-            // TODO: Return error if any
-            hd_tcp::write_ack(&mut client.stream, PacketType::Publish)?;
-            debug!("Publish ack complete");
-            Ok(EventBusClientConnectionState::WaitEvent)
-        }
-        p => {
-            error!(
-                "Received wrong packet of type {}, expected {}. Will disconnect",
-                get_packet_type(&p),
-                PacketType::Connect
-            );
-            Ok(EventBusClientConnectionState::Disconnect)
-        }
-    }
-}
-
-fn handle_state_send_event(client: &mut EventBusClientConnection) -> Result<EventBusClientConnectionState, HError> {
-    if let Some(subscription) = &mut client.subscription {
-        // TODO: When client sends unsubscribe, server shouldn't take forever
-        // to ack that.
-        let Ok(evt) = subscription.recv() else {
-            error!("Failed to receive a message for this subscription, disconnecting");
-            return Ok(EventBusClientConnectionState::Disconnect);
-        };
-
-        let packet = Packet::create_send_event_packet(evt);
-        hd_tcp::write_packet(&mut client.stream, &packet)?;
-        debug!("Published event to client, waiting for ack");
-
-        let ack_packet = hd_tcp::read_ack(&mut client.stream, &[PacketType::SendEvent, PacketType::Unsubscribe])?;
-        if packet::get_packet_type(&ack_packet) == PacketType::Unsubscribe {
-            info!("Subscriber unsubscribed");
-            return Ok(EventBusClientConnectionState::Command);
-        }
-
-        debug!("Publish ack received, message is settled");
-    }
-
-    Ok(EventBusClientConnectionState::SendEvent)
-}
-
-fn handle_state_wait_event(client: &mut EventBusClientConnection) -> Result<EventBusClientConnectionState, HError> {
-    match hd_tcp::read_packet(&mut client.stream)? {
-        Packet::Publish(publish_packet) => {
-            debug!("Received event from client, publishing to bus");
-            publish_packet_to_bus(client, publish_packet)?;
-
-            hd_tcp::write_ack(&mut client.stream, PacketType::Publish)?;
-            debug!("Ack receive event complete");
-            Ok(EventBusClientConnectionState::WaitEvent)
-        }
-        p => {
-            error!(
-                "Received wrong packet of type {}, expected {}. Will disconnect",
-                get_packet_type(&p),
-                PacketType::Connect
-            );
-            Ok(EventBusClientConnectionState::Disconnect)
-        }
-    }
-}
-
-fn publish_packet_to_bus(client: &mut EventBusClientConnection, packet: PublishPacket) -> Result<(), HError> {
-    let data = String::from_utf8(packet.data_utf8)?;
-    let evt = EventData::new(packet.event_type, &data);
-    client.bus.publish(evt)?;
-    Ok(())
-}
-
-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::Command => "COMMAND",
-            EventBusClientConnectionState::SendEvent => "SEND_EVENT",
-            EventBusClientConnectionState::WaitEvent => "WAIT_EVENT",
-            EventBusClientConnectionState::Disconnect => "DISCONNECT",
-        };
-
-        write!(f, "{}", to_write)
-    }
-}