WIP

crates/hd-lib/Cargo.toml

@@ -0,0 +1,7 @@
+[package]
+name = "hd-lib"
+version = "0.1.0"
+edition = "2024"
+
+[dependencies]
+tracing = { workspace = true }

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

@@ -0,0 +1,218 @@
+mod packet;
+
+use tracing::{info, info_span};
+
+use crate::error::HError;
+use std::{
+    sync::{
+        Arc, RwLock,
+        mpsc::{Receiver, Sender, SyncSender, channel, sync_channel},
+    },
+    thread::{self, JoinHandle},
+};
+
+pub type EventId = u64;
+pub type EventType = u8;
+pub type SubscriptionId = u64;
+pub type Cursor = u64;
+
+pub struct Event {
+    pub id: EventId,
+    pub data: EventData,
+}
+
+pub struct EventData {
+    pub event_type: EventType,
+    pub data_utf8: Vec<u8>,
+}
+
+impl EventData {
+    pub fn new(event_type: u8, data: &str) -> Self {
+        let data_utf8 = data.as_bytes();
+        Self {
+            event_type,
+            data_utf8: data_utf8.into(),
+        }
+    }
+}
+
+#[derive(Clone)]
+pub struct EventBus {
+    inner: Arc<RwLock<EventBusInner>>,
+}
+
+struct EventBusInner {
+    next_event_id: EventId,
+    events: Vec<Arc<Event>>,
+
+    subscription_handles: Vec<SubscriptionHandle>,
+    next_subscription_id: SubscriptionId,
+}
+
+impl EventBusInner {
+    fn new() -> Self {
+        Self {
+            next_event_id: 0,
+            next_subscription_id: 0,
+            events: Vec::new(),
+            subscription_handles: Vec::new(),
+        }
+    }
+}
+
+struct SubscriptionHandle {
+    id: SubscriptionId,
+    new_event_signal: Sender<u64>,
+    join_handle: JoinHandle<Result<(), HError>>,
+}
+
+pub struct PollResult {
+    pub events: Vec<Arc<Event>>,
+    // None means that no events were returned
+    pub cursor_end: Cursor,
+}
+
+impl EventBus {
+    pub fn new() -> Self {
+        Self {
+            inner: Arc::new(RwLock::new(EventBusInner::new())),
+        }
+    }
+
+    pub fn publish(&self, evt: EventData) -> Result<(), HError> {
+        let mut bus = self.inner.write().map_err(|e| HError::BusLockPoisoned(e.to_string()))?;
+
+        let next_event_id = &mut bus.next_event_id;
+        let event = Event {
+            id: *next_event_id,
+            data: evt,
+        };
+
+        *next_event_id += 1;
+
+        let events = &mut bus.events;
+        events.push(Arc::new(event));
+
+        for s in &bus.subscription_handles {
+            if let Err(e) = s.new_event_signal.send(1) {
+                tracing::error!("Failed to notify {}", e);
+            }
+        }
+
+        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()))?;
+
+        for evt in &bus.events {
+            if evt.id >= cursor {
+                return Ok(evt.clone());
+            }
+        }
+
+        Err(HError::NoMoreEvents)
+    }
+
+    pub fn subscribe(&self) -> Result<Receiver<Arc<Event>>, HError> {
+        let mut bus = self.inner.write().map_err(|e| HError::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);
+
+        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());
+
+        let subscription_handle = SubscriptionHandle {
+            id: next_subscription_id,
+            new_event_signal: new_event_signal_sender,
+            join_handle: bg_thread_handle,
+        };
+
+        bus.next_subscription_id += 1;
+        bus.subscription_handles.push(subscription_handle);
+
+        Ok(event_recv)
+    }
+
+    pub fn unsubscribe(&self, id: SubscriptionId) -> Result<(), HError> {
+        let mut bus = self.inner.write().map_err(|e| HError::BusLockPoisoned(e.to_string()))?;
+
+        let mut found = None;
+        let mut index = 0;
+        for sub in &mut bus.subscription_handles {
+            if sub.id == id {
+                found = Some(index);
+                break;
+            }
+
+            index += 1;
+        }
+
+        match found {
+            None => Err(HError::SubscriptionNotFound(format!(
+                "Subscription with id {} is not found",
+                id
+            ))),
+            Some(i) => {
+                // This will drop the join handle of the background thread.
+                let subscription = bus.subscription_handles.swap_remove(i);
+                drop(subscription.join_handle);
+                Ok(())
+            }
+        }
+    }
+}
+
+pub struct Subscription {
+    id: SubscriptionId,
+    // Return events inclusive of the cursor on a poll
+    cursor: Cursor,
+    new_event_signal: Receiver<u64>,
+    client: SyncSender<Arc<Event>>,
+}
+
+impl Subscription {
+    pub fn new(
+        id: SubscriptionId,
+        cursor: Cursor,
+        new_event_signal: Receiver<u64>,
+        client: SyncSender<Arc<Event>>,
+    ) -> Self {
+        Self {
+            id,
+            cursor,
+            new_event_signal,
+            client,
+        }
+    }
+}
+
+fn start_delivery(mut subscription: Subscription, bus: EventBus) -> JoinHandle<Result<(), HError>> {
+    thread::spawn(move || {
+        loop {
+            let _scope = info_span!("subscription_delivery", subscription.id).entered();
+            loop {
+                if let Ok(evt) = bus.get_next_event(subscription.cursor) {
+                    info!("Delivering event {} ", evt.id);
+                    if let Err(e) = subscription.client.send(evt) {
+                        tracing::error!("Event delivery for id {} failed {}", subscription.id, e);
+                        return Ok(());
+                    }
+
+                    subscription.cursor += 1;
+                } else {
+                    break;
+                }
+            }
+
+            if subscription.new_event_signal.recv().is_err() {
+                return Ok(());
+            }
+        }
+    })
+}

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

@@ -0,0 +1,277 @@
+use std::{io::{Read, Write}, net::TcpStream};
+
+use crate::bus::EventData;
+
+enum PacketType {
+    Connect,
+    Unsubscribe,
+    Publish,
+    SendEvent,
+    Settle,
+    Ack,
+    Disconnect,
+}
+
+enum Packet {
+    Connect(ConnectPacket),
+    Unsubscribe,
+    Publish(PublishPacket),
+    SendEvent(SendEventPacket),
+    Settle(SettlePacket),
+    Ack(AckPacket),
+    Disconnect,
+}
+
+struct ConnectPacket {
+    cursor: u64,
+}
+
+struct PublishPacket {
+    event_type: u8,
+    data_utf8: Vec<u8>,
+}
+
+struct SendEventPacket {
+    event_id: u64,
+    event_type: u8,
+    data_utf8: Vec<u8>,
+}
+
+struct SettlePacket {
+    cursor: u64,
+}
+
+struct AckPacket {
+    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,
+        }
+    }
+
+    fn create_connect_packet(cursor: u64) -> Packet {
+        Packet::Connect(ConnectPacket { cursor })
+    }
+
+    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 {
+        Packet::Settle(SettlePacket { cursor })
+    }
+
+    fn create_ack_packet(packet_type: PacketType) -> Packet {
+        Packet::Ack(AckPacket { packet_type })
+    }
+
+    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 {
+    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)?;
+
+    let packet_type = get_packet_type_from_byte(header_buf[0])?;
+
+    match packet_type {
+        PacketType::Connect => {
+            let cursor = read_u64(stream)?;
+            Ok(Packet::Connect(ConnectPacket { cursor }))
+        }
+        PacketType::Unsubscribe => Ok(Packet::Unsubscribe),
+        PacketType::Publish => todo!(),
+        PacketType::SendEvent => todo!(),
+        PacketType::Settle => {
+            let cursor = read_u64(stream)?;
+            Ok(Packet::Settle(SettlePacket { cursor }))
+        }
+        PacketType::Ack => {
+            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> {
+    let packet_type = packet.get_packet_type();
+    let data_frame_length = calculate_data_frame_length(&packet);
+
+    let mut header_frame = [0; 3];
+    write_header_frame(&mut header_frame, packet_type, data_frame_length);
+
+    if data_frame_length > 0 {
+        let mut data_frame = vec![0; data_frame_length];
+        write_data_frame(&mut data_frame, packet);
+    }
+
+    // RESUME FROM HERE
+
+    Ok(())
+}
+
+fn write_header_frame(header_frame: &mut [u8; 3], 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]);
+    }
+}
+
+fn write_data_frame(data_frame: &mut [u8], packet: &Packet) {
+    match packet {
+        Packet::Connect(connect_packet) => {
+            write_u64(connect_packet.cursor, data_frame);
+        }
+        Packet::Publish(publish_packet) => {
+            write_event_data(publish_packet.event_type, &publish_packet.data_utf8, data_frame);
+        }
+        Packet::SendEvent(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..],
+            );
+        }
+        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);
+        }
+        _ => {}
+    }
+}
+
+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]);
+    let available_length = buf.len() - 3;
+    buf[3..].copy_from_slice(&data_utf8[..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);
+}
+
+fn read_u64(stream: &mut TcpStream) -> Result<u64, PacketError> {
+    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);
+}
+
+fn calculate_data_frame_length(packet: &Packet) -> usize {
+    let mut ret = match packet {
+        Packet::Connect(_) => 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(),
+        Packet::Settle(_) => 8,
+        Packet::Ack(_) => 1,
+        Packet::Disconnect => 0,
+    };
+
+    // Normalize to 2 bytes
+    if ret > 0xffff {
+        ret = 0xffff;
+    }
+
+    ret
+}

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

@@ -0,0 +1,22 @@
+use std::fmt::Display;
+
+#[derive(Debug)]
+pub enum HError {
+    BusLockPoisoned(String),
+    SubscriptionNotFound(String),
+    TcpSocketBindError(std::io::Error),
+    TcpPeerError(std::io::Error),
+    NoMoreEvents,
+}
+
+impl Display for HError {
+    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::TcpSocketBindError(error) => write!(f, "{}", error),
+            HError::TcpPeerError(error) => write!(f, "{}", error),
+            HError::NoMoreEvents => write!(f, "{}", "No more events in the bus for now"),
+        }
+    }
+}

crates/hd-lib/src/lib.rs

@@ -0,0 +1,3 @@
+pub mod error;
+pub mod thread_pool;
+pub mod bus;

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

@@ -0,0 +1,64 @@
+use std::{
+    sync::{
+        Arc, Mutex,
+        mpsc::{Receiver, SyncSender, sync_channel},
+    },
+    thread::{self, JoinHandle},
+};
+
+use tracing::{Level, info, span};
+
+use crate::error::HError;
+
+pub struct ThreadPool {
+    workers: Vec<Worker>,
+    tx: SyncSender<Job>,
+}
+
+struct Worker {
+    handle: JoinHandle<()>,
+    id: usize,
+}
+
+struct Job {
+    f: Box<dyn FnOnce() -> Result<(), HError> + Send>,
+}
+
+impl ThreadPool {
+    pub fn new(max: usize) -> Self {
+        let mut workers = Vec::new();
+        let (tx, rx) = sync_channel(20);
+        let rx = Arc::new(Mutex::new(rx));
+        for i in 0..max {
+            workers.push(ThreadPool::worker_thread(i, rx.clone()));
+        }
+
+        Self { workers, tx }
+    }
+
+    pub fn execute<F>(&self, f: F) -> Result<(), HError>
+    where
+        F: FnOnce() -> Result<(), HError>,
+        F: Send + 'static,
+    {
+        let job = Job { f: Box::new(f) };
+        self.tx.send(job).unwrap();
+
+        Ok(())
+    }
+
+    fn worker_thread(id: usize, rx: Arc<Mutex<Receiver<Job>>>) -> Worker {
+        let handle = thread::spawn(move || {
+            let s = span!(Level::INFO, "worker_thread", id = 1);
+            let _scope = s.enter();
+            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");
+            }
+        });
+
+        Worker { id, handle }
+    }
+}