Initial protocol

Cargo.lock

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

crates/hd-client/Cargo.toml

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

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

@@ -1,6 +1,10 @@
 use hd_client::{EventBusClient, EventBusClientError};
 
 fn main() -> Result<(), EventBusClientError> {
+    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)?;
     client.subscribe()

crates/hd-client/src/lib.rs

@@ -1,11 +1,22 @@
-use std::{
-    io::{Read, Write},
-    net::TcpStream,
+use std::net::TcpStream;
+
+use hd_lib::{
+    bus::packet::{self, Packet, PacketType},
+    error::HError,
+    hd_tcp,
 };
+use tracing::{debug, info};
 
 #[derive(Debug)]
 pub enum EventBusClientError {
     TcpStreamError(std::io::Error),
+    HError(HError),
+}
+
+impl From<HError> for EventBusClientError {
+    fn from(value: HError) -> Self {
+        Self::HError(value)
+    }
 }
 
 pub struct EventBusClient {
@@ -14,43 +25,28 @@ pub struct EventBusClient {
 
 impl EventBusClient {
     pub fn start(addr: &'static str) -> Result<Self, EventBusClientError> {
-        let stream = TcpStream::connect(addr).map_err(EventBusClientError::TcpStreamError)?;
-        println!("Event bus client connected to {}", addr);
+        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])?;
 
+        info!("Event bus client connected to {}", addr);
         Ok(Self { tcp_stream: stream })
     }
 
     pub fn subscribe(&mut self) -> Result<(), EventBusClientError> {
-        let bytes = [1];
-        self.tcp_stream
-            .write_all(&bytes)
-            .map_err(EventBusClientError::TcpStreamError)?;
+        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])?;
 
-        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 {
-                return Ok(());
-            };
-
-            if bytes_read == 0 {
-                eprintln!("Connection closed");
+            let packet = hd_tcp::read_packet(&mut self.tcp_stream)?;
+            if let Packet::Disconnect = packet {
+                info!("Disconnecting from the server");
                 return Ok(());
             }
 
-            println!("MESSAGE BEGIN");
-            for b in &buf[0..bytes_read] {
-                print!("[{}] ", b);
-            }
-            println!("MESSAGE END");
-
-            // ACK
-            self.tcp_stream
-                .write_all(&bytes)
-                .map_err(EventBusClientError::TcpStreamError)?;
-
-            self.tcp_stream.flush().map_err(EventBusClientError::TcpStreamError)?;
+            debug!("Received packet ... {}", packet::get_packet_type(&packet));
+            hd_tcp::write_ack(&mut self.tcp_stream, PacketType::SendEvent)?;
         }
     }
 }

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

@@ -1,6 +1,6 @@
-mod packet;
+pub mod packet;
 
-use tracing::{info, info_span};
+use tracing::{debug_span, info, info_span};
 
 use crate::error::HError;
 use std::{
@@ -95,6 +95,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);
             }
         }
@@ -195,7 +196,7 @@ impl Subscription {
 fn start_delivery(mut subscription: Subscription, bus: EventBus) -> JoinHandle<Result<(), HError>> {
     thread::spawn(move || {
         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,12 +1,15 @@
-use std::{
-    io::{Read, Write},
-    net::TcpStream,
-};
+use std::{fmt::Display, sync::Arc};
 
-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,
+    Subscribe,
     Unsubscribe,
     Publish,
     SendEvent,
@@ -15,8 +18,26 @@ enum PacketType {
     Disconnect,
 }
 
-enum Packet {
-    Connect(ConnectPacket),
+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::Disconnect => "DISCONNECT",
+        };
+
+        write!(f, "{}", to_write)
+    }
+}
+
+pub enum Packet {
+    Connect,
+    Subscribe(SubscribePacket),
     Unsubscribe,
     Publish(PublishPacket),
     SendEvent(SendEventPacket),
@@ -25,186 +46,183 @@ enum Packet {
     Disconnect,
 }
 
-struct ConnectPacket {
+pub struct SubscribePacket {
     cursor: u64,
 }
 
-struct PublishPacket {
-    event_type: u8,
-    data_utf8: Vec<u8>,
+pub struct PublishPacket {
+    pub event_type: u8,
+    pub data_utf8: Vec<u8>,
 }
 
-struct SendEventPacket {
+pub struct SendEventPacket {
     event_id: u64,
     event_type: u8,
     data_utf8: Vec<u8>,
 }
 
-struct SettlePacket {
+pub struct SettlePacket {
     cursor: u64,
 }
 
-struct AckPacket {
-    packet_type: PacketType,
+pub struct AckPacket {
+    pub packet_type: PacketType,
 }
 
 impl Packet {
-    fn get_packet_type(&self) -> PacketType {
-        match self {
-            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,
-        }
+    pub fn create_connect_packet() -> Packet {
+        Packet::Connect
     }
 
-    fn create_connect_packet(cursor: u64) -> Packet {
-        Packet::Connect(ConnectPacket { cursor })
+    pub fn create_subscribe_packet(cursor: u64) -> Packet {
+        Packet::Subscribe(SubscribePacket { cursor })
     }
 
-    fn create_unsubscribe_packet() -> Packet {
+    pub fn create_unsubscribe_packet() -> Packet {
         Packet::Unsubscribe
     }
 
-    fn create_publish_packet(evt: EventData) -> Packet {
+    // TODO: Avoid copying data_utf8
+    pub 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_send_event_packet(evt: Arc<Event>) -> Packet {
+        Packet::SendEvent(SendEventPacket {
+            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_ack_packet(packet_type: PacketType) -> Packet {
         Packet::Ack(AckPacket { packet_type })
     }
 
-    fn create_disconnect_packet() -> Packet {
+    pub fn create_disconnect_packet() -> Packet {
         Packet::Disconnect
     }
 }
 
-struct PacketFrame {
-    header: HeaderFrame,
-    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 {
+#[derive(Debug)]
+pub enum PacketError {
     WrongPacketType(u8),
     PacketProtocolError,
-    TcpStreamError(std::io::Error),
 }
 
-fn read_packet(stream: &mut TcpStream) -> Result<Packet, PacketError> {
-    let mut header_buf = vec![0; 3];
-    stream
-        .read_exact(&mut header_buf)
-        .map_err(PacketError::TcpStreamError)?;
+impl Display for PacketError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        let to_write = match self {
+            PacketError::WrongPacketType(p) => &format!("WRONG_PACKET_TYPE({})", p),
+            PacketError::PacketProtocolError => "PACKET_PROTOCOL_ERROR",
+        };
 
-    let packet_type = get_packet_type_from_byte(header_buf[0])?;
+        write!(f, "{}", to_write)
+    }
+}
 
-    match packet_type {
-        PacketType::Connect => {
-            let cursor = read_u64(stream)?;
-            Ok(Packet::Connect(ConnectPacket { cursor }))
+pub struct ReadPacketState {
+    pub required_buffer_size: usize,
+    packet_type: PacketType,
+}
+
+pub 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 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::Unsubscribe => Ok(Packet::Unsubscribe),
         PacketType::Publish => {
-            let mut header = vec![0u8; 3];
-            stream.read_exact(&mut header).map_err(PacketError::TcpStreamError)?;
+            let event_type = buf[0];
+            let event_len = read_u16(&buf[1..3])?;
 
-            let event_type = header[0];
-
-            let len = usize::from_le_bytes([header[1], header[2], 0, 0, 0, 0, 0, 0]);
-            let mut data_utf8 = vec![0u8; len];
-            stream.read_exact(&mut data_utf8).map_err(PacketError::TcpStreamError)?;
-
-            let publish_packet = PublishPacket { event_type, data_utf8 };
+            // TODO: Share underlying bytes without copying
+            let publish_packet = PublishPacket {
+                event_type,
+                data_utf8: buf[3..3 + event_len].to_vec(),
+            };
             Ok(Packet::Publish(publish_packet))
         }
         PacketType::SendEvent => {
-            let mut header = vec![0u8; 11];
-            stream.read_exact(&mut header).map_err(PacketError::TcpStreamError)?;
-
-            let mut le_bytes = [0u8; 8];
-            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 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 {
                 event_id,
                 event_type,
-                data_utf8,
+                data_utf8: buf[11..11 + event_len].to_vec(),
             };
             Ok(Packet::SendEvent(send_event_packet))
         }
         PacketType::Settle => {
-            let cursor = read_u64(stream)?;
+            let cursor = read_u64(&buf[0..8])?;
             Ok(Packet::Settle(SettlePacket { cursor }))
         }
         PacketType::Ack => {
-            let packet_type = get_packet_type_from_byte(read_byte(stream)?)?;
+            let packet_type = get_packet_type_from_byte(buf[0])?;
             Ok(Packet::Ack(AckPacket { packet_type }))
         }
         PacketType::Disconnect => Ok(Packet::Disconnect),
     }
 }
 
-fn write_packet(stream: &mut TcpStream, packet: &Packet) -> Result<(), PacketError> {
-    let packet_type = packet.get_packet_type();
+pub struct WritePacketState {
+    pub required_buffer_size: usize,
+    data_frame_length: usize,
+    packet_type: PacketType,
+}
+
+pub 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];
-    write_header_frame(&mut buf, packet_type, data_frame_length);
-
-    if data_frame_length > 0 {
+pub 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);
     }
 
-    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) => {
+        Packet::Subscribe(connect_packet) => {
             write_u64(connect_packet.cursor, data_frame);
         }
         Packet::Publish(publish_packet) => {
@@ -215,7 +233,7 @@ fn write_data_frame(data_frame: &mut [u8], packet: &Packet) {
             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) => {
@@ -230,49 +248,79 @@ fn write_data_frame(data_frame: &mut [u8], packet: &Packet) {
 
 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]) {
+    if length > 0xffff {
+        length = 0xffff;
+    }
+
+    let bytes = length.to_le_bytes();
+    buf[0..2].copy_from_slice(&bytes[0..2]);
+}
+
+fn read_u64(buf: &[u8]) -> Result<u64, PacketError> {
     let mut le_bytes = [0u8; 8];
-    stream.read_exact(&mut le_bytes).map_err(PacketError::TcpStreamError)?;
+    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_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 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::Subscribe(_) => PacketType::Subscribe,
+        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 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,
+        PacketType::Subscribe => 1,
+        PacketType::Unsubscribe => 2,
+        PacketType::Publish => 3,
+        PacketType::SendEvent => 4,
+        PacketType::Settle => 5,
+        PacketType::Ack => 6,
+        PacketType::Disconnect => 7,
     }
 }
 
 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),
+        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,
     };
 
@@ -283,18 +331,10 @@ fn get_packet_type_from_byte(byte: u8) -> Result<PacketType, PacketError> {
     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);
-}
-
 fn calculate_data_frame_length(packet: &Packet) -> usize {
     let mut ret = match packet {
-        Packet::Connect(_) => 8,
+        Packet::Connect => 0,
+        Packet::Subscribe(_) => 8,
         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(),

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

@@ -1,12 +1,17 @@
-use std::fmt::Display;
+use std::{fmt::Display, string::FromUtf8Error};
+
+use crate::bus::packet::PacketError;
 
 #[derive(Debug)]
 pub enum HError {
     BusLockPoisoned(String),
     SubscriptionNotFound(String),
-    TcpSocketBindError(std::io::Error),
-    TcpPeerError(std::io::Error),
+    IoError(std::io::Error),
+    TcpPacketError(PacketError),
+    Utf8FormatError(FromUtf8Error),
     NoMoreEvents,
+    ProtocolError,
+    PeerDisconnect,
 }
 
 impl Display for HError {
@@ -14,9 +19,30 @@ impl Display for HError {
         match self {
             HError::BusLockPoisoned(s) => write!(f, "{}", s),
             HError::SubscriptionNotFound(s) => write!(f, "{}", s),
-            HError::TcpSocketBindError(error) => write!(f, "{}", error),
-            HError::TcpPeerError(error) => write!(f, "{}", error),
+            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"),
         }
     }
 }
+
+impl From<std::io::Error> for HError {
+    fn from(value: std::io::Error) -> Self {
+        HError::IoError(value)
+    }
+}
+
+impl From<PacketError> for HError {
+    fn from(value: PacketError) -> Self {
+        HError::TcpPacketError(value)
+    }
+}
+
+impl From<FromUtf8Error> for HError {
+    fn from(value: FromUtf8Error) -> Self {
+        HError::Utf8FormatError(value)
+    }
+}

crates/hd-lib/src/lib.rs

@@ -1,3 +1,62 @@
+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, PacketType},
+        error::HError,
+    };
+
+    pub fn read_packet(stream: &mut TcpStream) -> Result<Packet, HError> {
+        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)?;
+        }
+        let packet = packet::read_packet(&buf[0..], read_state)?;
+
+        if let Packet::Disconnect = packet {
+            return Err(HError::PeerDisconnect);
+        };
+
+        Ok(packet)
+    }
+
+    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> {
+        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-server/src/bin/server.rs

@@ -10,14 +10,17 @@ fn main() -> Result<(), HError> {
         .init();
 
     let addr = "0.0.0.0:21368";
-    let _scope = info_span!("example_server").entered();
     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-server/src/lib.rs

@@ -1,15 +1,20 @@
 use std::{
-    io::{Read, Write},
     net::TcpStream,
     sync::{Arc, mpsc::Receiver},
     time::Duration,
 };
 
 use hd_lib::{
-    bus::{Event, EventBus, EventData}, error::HError, thread_pool::ThreadPool,
+    bus::{
+        Event, EventBus, EventData,
+        packet::{self, Packet, PacketType, PublishPacket, get_packet_type},
+    },
+    error::HError,
+    hd_tcp,
+    thread_pool::ThreadPool,
 };
 
-use tracing::{Span, info, info_span};
+use tracing::{debug, error, info, info_span};
 
 use std::net::TcpListener;
 
@@ -34,12 +39,12 @@ impl TcpEventBus {
 
 impl TcpEventBus {
     pub fn start(&self, addr: &'static str) -> Result<(), HError> {
-        let listener = TcpListener::bind(addr).map_err(HError::TcpSocketBindError)?;
-        info!("Heimdall listening at {}", addr);
+        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)?,
-                Err(e) => tracing::error!("TCP connection failed {}", e),
+                Err(e) => error!("TCP connection failed {}", e),
             }
         }
 
@@ -50,16 +55,15 @@ impl TcpEventBus {
 #[derive(Copy, Clone, PartialEq)]
 enum EventBusClientConnectionState {
     Connect,
-    ReadPrelude,
-    ReadCursor,
+    Command,
     SendEvent,
-    ReceiveAck,
+    WaitEvent,
     Disconnect,
 }
 
 struct EventBusClientConnection {
-    stream: TcpStream,
     pub addr: String,
+    stream: TcpStream,
     state: EventBusClientConnectionState,
     subscription: Option<Receiver<Arc<Event>>>,
     bus: EventBus,
@@ -72,9 +76,7 @@ impl EventBusClientConnection {
             .map(|a| a.to_string())
             .unwrap_or("unknown".to_string());
 
-        stream
-            .set_read_timeout(Some(Duration::from_secs(5)))
-            .map_err(HError::TcpPeerError)?;
+        stream.set_read_timeout(Some(Duration::from_secs(5)))?;
 
         let state = EventBusClientConnectionState::Connect;
 
@@ -92,12 +94,7 @@ pub fn handle_event_bus_client(thread_pool: &ThreadPool, bus: EventBus, stream:
     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();
+        let _scope = info_span!("handle_client", addr = client.addr,).entered();
 
         loop {
             if let EventBusClientConnectionState::Disconnect = client.state {
@@ -111,8 +108,6 @@ pub fn handle_event_bus_client(thread_pool: &ThreadPool, bus: EventBus, stream:
             if prev_state != client.state {
                 info!("State change {} -> {}", prev_state, client.state);
             }
-
-            Span::current().record("state", format!("{}", client.state));
         }
     })
 }
@@ -120,11 +115,10 @@ pub fn handle_event_bus_client(thread_pool: &ThreadPool, bus: EventBus, stream:
 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::Command => handle_state_command(client),
         EventBusClientConnectionState::SendEvent => handle_state_send_event(client),
-        EventBusClientConnectionState::ReceiveAck => todo!(),
-        _ => Ok(EventBusClientConnectionState::Disconnect),
+        EventBusClientConnectionState::WaitEvent => handle_state_wait_event(client),
+        EventBusClientConnectionState::Disconnect => Ok(EventBusClientConnectionState::Disconnect),
     }?;
 
     client.state = next_state;
@@ -132,74 +126,121 @@ fn handle_event_bus_client_state(client: &mut EventBusClientConnection) -> Resul
 }
 
 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);
-    }
+    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 {
-        loop {
+        // 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 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;
+        let packet = Packet::create_send_event_packet(evt);
+        hd_tcp::write_packet(&mut client.stream, &packet)?;
+        debug!("Published event to client, waiting for ack");
 
-            for i in 17..buf_length {
-                buf[i] = evt.data.data_utf8[i - 17];
+        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);
         }
 
-            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)?;
-        }
+        debug!("Publish ack received, message is settled");
     }
 
-    Ok(EventBusClientConnectionState::Disconnect)
+    Ok(EventBusClientConnectionState::SendEvent)
 }
 
-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];
+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::ReadPrelude => "READ_PRELUDE",
-            EventBusClientConnectionState::ReadCursor => "READ_CURSOR",
+            EventBusClientConnectionState::Command => "COMMAND",
             EventBusClientConnectionState::SendEvent => "SEND_EVENT",
-            EventBusClientConnectionState::ReceiveAck => "RECEIVE_ACK",
+            EventBusClientConnectionState::WaitEvent => "WAIT_EVENT",
             EventBusClientConnectionState::Disconnect => "DISCONNECT",
         };