From 0c3b9cce91821644f99fde34622079a7f0a345bd Mon Sep 17 00:00:00 2001 From: cool-mist Date: Mon, 31 Aug 2026 23:13:15 +0530 Subject: [PATCH] WIP --- Cargo.lock | 4 + crates/hd-bus/src/bin/server.rs | 10 ++ crates/hd-bus/src/bus.rs | 213 +++++++++++++++++++++++++++++ crates/hd-bus/src/connection.rs | 164 ++++++++++++++++++++++ crates/hd-bus/src/error.rs | 22 +++ crates/hd-bus/src/lib.rs | 192 ++++---------------------- crates/hd-bus/src/thread_pool.rs | 60 ++++++++ crates/hd-client/Cargo.toml | 6 + crates/hd-client/src/bin/client.rs | 7 + crates/hd-client/src/lib.rs | 56 ++++++++ rustfmt.toml | 1 + 11 files changed, 571 insertions(+), 164 deletions(-) create mode 100644 crates/hd-bus/src/bin/server.rs create mode 100644 crates/hd-bus/src/bus.rs create mode 100644 crates/hd-bus/src/connection.rs create mode 100644 crates/hd-bus/src/error.rs create mode 100644 crates/hd-bus/src/thread_pool.rs create mode 100644 crates/hd-client/Cargo.toml create mode 100644 crates/hd-client/src/bin/client.rs create mode 100644 crates/hd-client/src/lib.rs create mode 100644 rustfmt.toml diff --git a/Cargo.lock b/Cargo.lock index 3837a10..8831ed0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5,3 +5,7 @@ version = 4 [[package]] name = "hd-bus" version = "0.1.0" + +[[package]] +name = "hd-client" +version = "0.1.0" diff --git a/crates/hd-bus/src/bin/server.rs b/crates/hd-bus/src/bin/server.rs new file mode 100644 index 0000000..f6ff119 --- /dev/null +++ b/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) +} diff --git a/crates/hd-bus/src/bus.rs b/crates/hd-bus/src/bus.rs new file mode 100644 index 0000000..f8c36e8 --- /dev/null +++ b/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, +} + +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>, +} + +struct EventBusInner { + next_event_id: EventId, + events: Vec>, + + subscription_handles: Vec, + 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, + join_handle: JoinHandle>, +} + +pub struct PollResult { + pub events: Vec>, + // 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, 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>, 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, + client: SyncSender>, +} + +impl Subscription { + pub fn new( + id: SubscriptionId, + cursor: Cursor, + new_event_signal: Receiver, + client: SyncSender>, + ) -> Self { + Self { + id, + cursor, + new_event_signal, + client, + } + } +} + +fn start_delivery(mut subscription: Subscription, bus: EventBus) -> JoinHandle> { + 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(()); + } + } + }) +} diff --git a/crates/hd-bus/src/connection.rs b/crates/hd-bus/src/connection.rs new file mode 100644 index 0000000..2008853 --- /dev/null +++ b/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>>, + bus: EventBus, +} + +impl EventBusClientConnection { + fn new(bus: EventBus, stream: TcpStream) -> Result { + 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 { + 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 { + 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) + } +} diff --git a/crates/hd-bus/src/error.rs b/crates/hd-bus/src/error.rs new file mode 100644 index 0000000..fa98d0e --- /dev/null +++ b/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"), + } + } +} diff --git a/crates/hd-bus/src/lib.rs b/crates/hd-bus/src/lib.rs index 78711d6..6ce72d6 100644 --- a/crates/hd-bus/src/lib.rs +++ b/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, -} - -// 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>, -} - -struct EventBusInner { - next_event_id: EventId, - events: Vec>, - subscriptions: RwLock, -} - -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, -} - -pub struct Subscription { - id: SubscriptionId, - // Return events exclusive of the cursor on a poll - cursor: EventId, -} - -pub struct PollResult { - pub events: Vec>, - 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 { - 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 { - 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() {} -} diff --git a/crates/hd-bus/src/thread_pool.rs b/crates/hd-bus/src/thread_pool.rs new file mode 100644 index 0000000..0fd6105 --- /dev/null +++ b/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, + tx: SyncSender, +} + +struct Worker { + handle: JoinHandle<()>, + id: usize, +} + +struct Job { + f: Box 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(&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>>) -> 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 } + } +} diff --git a/crates/hd-client/Cargo.toml b/crates/hd-client/Cargo.toml new file mode 100644 index 0000000..3622bd1 --- /dev/null +++ b/crates/hd-client/Cargo.toml @@ -0,0 +1,6 @@ +[package] +name = "hd-client" +version = "0.1.0" +edition = "2024" + +[dependencies] diff --git a/crates/hd-client/src/bin/client.rs b/crates/hd-client/src/bin/client.rs new file mode 100644 index 0000000..12e92ed --- /dev/null +++ b/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() +} diff --git a/crates/hd-client/src/lib.rs b/crates/hd-client/src/lib.rs new file mode 100644 index 0000000..b0dcf5e --- /dev/null +++ b/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 { + 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)?; + } + } +} diff --git a/rustfmt.toml b/rustfmt.toml new file mode 100644 index 0000000..7530651 --- /dev/null +++ b/rustfmt.toml @@ -0,0 +1 @@ +max_width = 120