WIP

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

@@ -0,0 +1,10 @@
+use hd_bus::{TcpEventBus, bus::EventData, error::HError};
+
+fn main() -> Result<(), HError> {
+    let server = TcpEventBus::new();
+    server.publish(EventData::new(1, "Hello"))?;
+    server.publish(EventData::new(1, "World"))?;
+
+    let addr = "0.0.0.0:21368";
+    server.start(addr)
+}

crates/hd-bus/src/bus.rs

@@ -0,0 +1,213 @@
+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) {
+                eprintln!("Failed to notify subscription with id {} - {}", s.id, 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 {
+            loop {
+                if let Ok(evt) = bus.get_next_event(subscription.cursor) {
+                    eprintln!("Delivering event {} to subscription {}", evt.id, subscription.id);
+                    if let Err(e) = subscription.client.send(evt) {
+                        eprintln!("Subscription 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-bus/src/connection.rs

@@ -0,0 +1,164 @@
+use std::{
+    io::{Read, Write},
+    net::TcpStream,
+    sync::{Arc, mpsc::Receiver},
+    time::Duration,
+};
+
+use crate::{
+    bus::{Event, EventBus},
+    error::HError,
+    thread_pool::ThreadPool,
+};
+
+#[derive(Copy, Clone, PartialEq)]
+enum EventBusClientConnectionState {
+    Connect,
+    ReadPrelude,
+    ReadCursor,
+    SendEvent,
+    ReceiveAck,
+    Disconnect,
+}
+
+struct EventBusClientConnection {
+    stream: TcpStream,
+    pub addr: String,
+    state: EventBusClientConnectionState,
+    subscription: Option<Receiver<Arc<Event>>>,
+    bus: EventBus,
+}
+
+impl EventBusClientConnection {
+    fn new(bus: EventBus, stream: TcpStream) -> Result<Self, HError> {
+        let addr = stream
+            .peer_addr()
+            .map(|a| a.to_string())
+            .unwrap_or("unknown".to_string());
+
+        stream
+            .set_read_timeout(Some(Duration::from_secs(5)))
+            .map_err(HError::TcpPeerError)?;
+
+        let state = EventBusClientConnectionState::Connect;
+
+        Ok(Self {
+            stream,
+            addr,
+            state,
+            bus,
+            subscription: None,
+        })
+    }
+}
+
+pub fn handle_event_bus_client(thread_pool: &ThreadPool, bus: EventBus, stream: TcpStream) -> Result<(), HError> {
+    thread_pool.execute(move || {
+        let mut client = EventBusClientConnection::new(bus, stream)?;
+
+        loop {
+            if let EventBusClientConnectionState::Disconnect = client.state {
+                eprintln!("[{}] Closing connection due to invalid input", client.addr);
+                return Ok(());
+            }
+
+            let prev_state = client.state;
+            handle_event_bus_client_state(&mut client)
+                .inspect_err(|e| eprintln!("[{}] Closing connection due to error {}", client.addr, e))?;
+            if prev_state != client.state {
+                println!("[{}] State change {} -> {}", client.addr, 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::ReadPrelude => todo!(),
+        EventBusClientConnectionState::ReadCursor => todo!(),
+        EventBusClientConnectionState::SendEvent => handle_state_send_event(client),
+        EventBusClientConnectionState::ReceiveAck => todo!(),
+        _ => Ok(EventBusClientConnectionState::Disconnect),
+    }?;
+
+    client.state = next_state;
+    Ok(())
+}
+
+fn handle_state_connect(client: &mut EventBusClientConnection) -> Result<EventBusClientConnectionState, HError> {
+    let mut buf = [0u8; 1];
+    client.stream.read_exact(&mut buf).map_err(HError::TcpPeerError)?;
+    if buf[0] == 1 {
+        let subscription = client.bus.subscribe()?;
+        client.subscription = Some(subscription);
+        return Ok(EventBusClientConnectionState::SendEvent);
+    }
+
+    Ok(EventBusClientConnectionState::Disconnect)
+}
+
+fn handle_state_send_event(client: &mut EventBusClientConnection) -> Result<EventBusClientConnectionState, HError> {
+    if let Some(subscription) = &mut client.subscription {
+        loop {
+            let Ok(evt) = subscription.recv() else {
+                return Ok(EventBusClientConnectionState::Disconnect);
+            };
+
+            let data_len = evt.data.data_utf8.len();
+            // [ ID  ] [TYPE] [ LEN ] [ DATA ]
+            // 0  to 7   8    9 to 15 . . . .
+            let buf_length = data_len + 17;
+            let mut buf = vec![0u8; buf_length];
+            write_le_bytes(evt.id, &mut buf[0..8]);
+            write_le_bytes(data_len as u64, &mut buf[9..17]);
+            buf[8] = evt.data.event_type;
+
+            for i in 17..buf_length {
+                buf[i] = evt.data.data_utf8[i - 17];
+            }
+
+            print!("Sending message : ");
+            for i in &buf {
+                print!("{} ", i);
+            }
+            println!("End");
+
+            let stream = &mut client.stream;
+            stream.write(&mut buf).map_err(HError::TcpPeerError)?;
+            stream.flush().map_err(HError::TcpPeerError)?;
+
+            let mut ack = vec![0];
+            stream.read(&mut ack).map_err(HError::TcpPeerError)?;
+        }
+    }
+
+    Ok(EventBusClientConnectionState::Disconnect)
+}
+
+fn write_le_bytes(v64: u64, buf: &mut [u8]) {
+    let le_bytes = v64.to_le_bytes();
+    buf[0] = le_bytes[0];
+    buf[1] = le_bytes[1];
+    buf[2] = le_bytes[2];
+    buf[3] = le_bytes[3];
+    buf[4] = le_bytes[4];
+    buf[5] = le_bytes[5];
+    buf[6] = le_bytes[6];
+    buf[7] = le_bytes[7];
+}
+
+impl std::fmt::Display for EventBusClientConnectionState {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        let to_write = match self {
+            EventBusClientConnectionState::Connect => "CONNECT",
+            EventBusClientConnectionState::ReadPrelude => "READ_PRELUDE",
+            EventBusClientConnectionState::ReadCursor => "READ_CURSOR",
+            EventBusClientConnectionState::SendEvent => "SEND_EVENT",
+            EventBusClientConnectionState::ReceiveAck => "RECEIVE_ACK",
+            EventBusClientConnectionState::Disconnect => "DISCONNECT",
+        };
+
+        write!(f, "{}", to_write)
+    }
+}

crates/hd-bus/src/error.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-bus/src/lib.rs

@@ -1,182 +1,46 @@
-use std::sync::{Arc, RwLock};
+pub mod bus;
+mod connection;
+pub mod error;
+mod thread_pool;
 
-pub type EventId = u64;
-pub type SubscriptionId = u64;
+use crate::{
+    bus::{EventBus, EventData},
+    connection::handle_event_bus_client,
+    error::HError,
+    thread_pool::ThreadPool,
+};
+use std::net::TcpListener;
 
-pub enum HError {
-    BusLockPoisoned(String),
+pub struct TcpEventBus {
+    bus: EventBus,
+    pool: ThreadPool,
 }
 
-pub struct Event {
-    id: EventId,
-    pub data: EventData,
-}
-
-pub struct EventData {
-    pub subject: String,
-    pub data_utf8: Vec<u8>,
-}
-
-// Add Event              event_bus scope W
-// Add subscription       event_bus scope W
-// Poll subscription      event_bus scope R + subscription scope R
-// Ack subscription       event_bus scope R + subscription scope W
-// Remove subscription    event_bus scope W
-// Cleanup Event          event_bus scope W
-pub struct EventBus {
-    inner: Arc<RwLock<EventBusInner>>,
-}
-
-struct EventBusInner {
-    next_event_id: EventId,
-    events: Vec<Arc<Event>>,
-    subscriptions: RwLock<SubscriptionsInner>,
-}
-
-impl EventBusInner {
-    fn new() -> Self {
-        let subscriptions = SubscriptionsInner {
-            next_subscription_id: 0,
-            subscriptions: Vec::new(),
-        };
-
-        Self {
-            next_event_id: 0,
-            events: Vec::new(),
-            subscriptions: RwLock::new(subscriptions),
-        }
-    }
-}
-
-struct SubscriptionsInner {
-    next_subscription_id: SubscriptionId,
-    subscriptions: Vec<Subscription>,
-}
-
-pub struct Subscription {
-    id: SubscriptionId,
-    // Return events exclusive of the cursor on a poll
-    cursor: EventId,
-}
-
-pub struct PollResult {
-    pub events: Vec<Arc<Event>>,
-    pub cursor_end: EventId,
-}
-
-impl EventBus {
+impl TcpEventBus {
     pub fn new() -> Self {
+        let pool = ThreadPool::new(10);
         Self {
-            inner: Arc::new(RwLock::new(EventBusInner::new())),
+            bus: EventBus::new(),
+            pool,
         }
     }
 
     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));
-        Ok(())
+        self.bus.publish(evt)
     }
+}
 
-    pub fn subscribe(&self) -> Result<SubscriptionId, HError> {
-        let bus = self
-            .inner
-            .write()
-            .map_err(|e| HError::BusLockPoisoned(e.to_string()))?;
-
-        let subscriptions = &mut bus
-            .subscriptions
-            .write()
-            .map_err(|e| HError::BusLockPoisoned(e.to_string()))?;
-
-        let next_subscription_id = &mut subscriptions.next_subscription_id;
-        let subscription = Subscription {
-            id: *next_subscription_id,
-            cursor: 0,
-        };
-
-        *next_subscription_id += 1;
-
-        let ret = subscription.id;
-        subscriptions.subscriptions.push(subscription);
-        Ok(ret)
-    }
-
-    pub fn poll(&self, id: SubscriptionId, max: usize) -> Result<PollResult, HError> {
-        let bus = self
-            .inner
-            .read()
-            .map_err(|e| HError::BusLockPoisoned(e.to_string()))?;
-
-        let mut cursor = 0;
-
-        {
-            let subscriptions = &mut bus
-                .subscriptions
-                .read()
-                .map_err(|e| HError::BusLockPoisoned(e.to_string()))?;
-
-            for sub in &subscriptions.subscriptions {
-                if sub.id == id {
-                    cursor = sub.cursor;
-                    break;
-                }
-            }
-        }
-
-        let mut events = Vec::new();
-        let mut cursor_end = cursor;
-        for evt in &bus.events {
-            if evt.id >= cursor {
-                events.push(evt.clone());
-                cursor_end = evt.id;
-                if events.len() >= max {
-                    break;
-                }
-            }
-        }
-
-        Ok(PollResult { events, cursor_end })
-    }
-
-    pub fn ack(&self, id: SubscriptionId, cursor: EventId) -> Result<(), HError> {
-        let bus = self
-            .inner
-            .read()
-            .map_err(|e| HError::BusLockPoisoned(e.to_string()))?;
-
-        {
-            let subscriptions = &mut bus
-                .subscriptions
-                .write()
-                .map_err(|e| HError::BusLockPoisoned(e.to_string()))?;
-
-            for sub in &mut subscriptions.subscriptions {
-                if sub.id == id {
-                    sub.cursor = cursor;
-                    break;
-                }
+impl TcpEventBus {
+    pub fn start(&self, addr: &'static str) -> Result<(), HError> {
+        let listener = TcpListener::bind(addr).map_err(HError::TcpSocketBindError)?;
+        println!("EventBus listening for TCP packets at {}", addr);
+        for stream in listener.incoming() {
+            match stream {
+                Ok(stream) => handle_event_bus_client(&self.pool, self.bus.clone(), stream)?,
+                Err(e) => eprintln!("{}", e),
             }
         }
 
         Ok(())
     }
 }
-
-#[cfg(test)]
-mod tests {
-    #[test]
-    fn it_works() {}
-}

crates/hd-bus/src/thread_pool.rs

@@ -0,0 +1,60 @@
+use std::{
+    sync::{
+        Arc, Mutex,
+        mpsc::{Receiver, SyncSender, sync_channel},
+    },
+    thread::{self, JoinHandle},
+};
+
+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 || {
+            loop {
+                let job = rx.lock().unwrap().recv().unwrap(); // Lock is released here
+                println!("Worker [{}] received a new job, executing", id);
+                (job.f)().unwrap(); // TODO: Panic recovery
+                println!("Worker [{}] finished executing job", id);
+            }
+        });
+
+        Worker { id, handle }
+    }
+}

crates/hd-client/Cargo.toml

@@ -0,0 +1,6 @@
+[package]
+name = "hd-client"
+version = "0.1.0"
+edition = "2024"
+
+[dependencies]

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

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

crates/hd-client/src/lib.rs

@@ -0,0 +1,56 @@
+use std::{
+    io::{Read, Write},
+    net::TcpStream,
+};
+
+#[derive(Debug)]
+pub enum EventBusClientError {
+    TcpStreamError(std::io::Error),
+}
+
+pub struct EventBusClient {
+    tcp_stream: TcpStream,
+}
+
+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);
+
+        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)?;
+
+        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");
+                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)?;
+        }
+    }
+}