diff --git a/crates/hd-client/Cargo.toml b/crates/hd-client/Cargo.toml index 875005f..4f8dae3 100644 --- a/crates/hd-client/Cargo.toml +++ b/crates/hd-client/Cargo.toml @@ -5,5 +5,4 @@ edition = "2024" [dependencies] tracing = { workspace = true } -hd-lib = { path = "../hd-lib" } -tracing-subscriber = { version = "0.3", features = ["env-filter", "fmt"] } +hd-lib = {path = "../hd-lib"} diff --git a/crates/hd-client/src/lib.rs b/crates/hd-client/src/lib.rs index e270a2b..ea5a463 100644 --- a/crates/hd-client/src/lib.rs +++ b/crates/hd-client/src/lib.rs @@ -1,43 +1,42 @@ + use std::net::TcpStream; use hd_lib::{ - bus::packet::{self, Packet, PacketType}, - error::HError, + bus::packet::{self, Packet}, + error::HdError, hd_tcp, }; use tracing::{debug, info}; #[derive(Debug)] -pub enum EventBusClientError { +pub enum HdClientError { TcpStreamError(std::io::Error), - HError(HError), + HError(HdError), } -impl From for EventBusClientError { - fn from(value: HError) -> Self { +impl From for HdClientError { + fn from(value: HdError) -> Self { Self::HError(value) } } -pub struct EventBusClient { +pub struct HdClient { tcp_stream: TcpStream, } -impl EventBusClient { - pub fn start(addr: &'static str) -> Result { - let mut stream = TcpStream::connect(addr).map_err(EventBusClientError::TcpStreamError)?; - hd_tcp::write_packet(&mut stream, &Packet::create_connect_packet())?; - hd_tcp::read_ack(&mut stream, &[PacketType::Connect])?; - +impl HdClient { + pub fn start(addr: &'static str) -> Result { + let stream = TcpStream::connect(addr).map_err(HdClientError::TcpStreamError)?; info!("Event bus client connected to {}", addr); Ok(Self { tcp_stream: stream }) } - pub fn subscribe(&mut self) -> Result<(), EventBusClientError> { + pub fn subscribe(&mut self) -> Result<(), HdClientError> { debug!("Going to subscribe"); - hd_tcp::write_packet(&mut self.tcp_stream, &Packet::create_subscribe_packet(0))?; - hd_tcp::read_ack(&mut self.tcp_stream, &[PacketType::Subscribe])?; - + hd_tcp::write_packet( + &mut self.tcp_stream, + &Packet::create_connect_packet(packet::ConnectMode::Subscribe), + )?; loop { let packet = hd_tcp::read_packet(&mut self.tcp_stream)?; if let Packet::Disconnect = packet { @@ -46,7 +45,7 @@ impl EventBusClient { } debug!("Received packet ... {}", packet::get_packet_type(&packet)); - hd_tcp::write_ack(&mut self.tcp_stream, PacketType::SendEvent)?; + hd_tcp::write_packet(&mut self.tcp_stream, &Packet::create_settle_packet(1))?; } } } diff --git a/crates/hd-examples/Cargo.toml b/crates/hd-examples/Cargo.toml new file mode 100644 index 0000000..e775422 --- /dev/null +++ b/crates/hd-examples/Cargo.toml @@ -0,0 +1,11 @@ +[package] +name = "hd-examples" +version = "0.1.0" +edition = "2024" + +[dependencies] +tracing = { workspace = true } +hd-lib = { path = "../hd-lib" } +hd-client = { path = "../hd-client" } +hd-server = { path = "../hd-server" } +tracing-subscriber = { version = "0.3", features = ["env-filter", "fmt"] } diff --git a/crates/hd-client/src/bin/client.rs b/crates/hd-examples/src/bin/client.rs similarity index 54% rename from crates/hd-client/src/bin/client.rs rename to crates/hd-examples/src/bin/client.rs index 228baf1..68cddf9 100644 --- a/crates/hd-client/src/bin/client.rs +++ b/crates/hd-examples/src/bin/client.rs @@ -1,11 +1,11 @@ -use hd_client::{EventBusClient, EventBusClientError}; +use hd_client::{HdClient, HdClientError}; -fn main() -> Result<(), EventBusClientError> { +fn main() -> Result<(), HdClientError> { tracing_subscriber::fmt() .with_env_filter(tracing_subscriber::EnvFilter::from_default_env()) .init(); let addr = "0.0.0.0:21368"; - let mut client = EventBusClient::start(addr)?; + let mut client = HdClient::start(addr)?; client.subscribe() } diff --git a/crates/hd-server/src/bin/server.rs b/crates/hd-examples/src/bin/server.rs similarity index 81% rename from crates/hd-server/src/bin/server.rs rename to crates/hd-examples/src/bin/server.rs index 08a2801..a2156f0 100644 --- a/crates/hd-server/src/bin/server.rs +++ b/crates/hd-examples/src/bin/server.rs @@ -1,16 +1,16 @@ use std::{sync::Arc, thread, time::Duration}; -use hd_lib::{bus::EventData, error::HError}; -use hd_server::TcpEventBus; +use hd_lib::{bus::EventData, error::HdError}; +use hd_server::HdBus; use tracing::info_span; -fn main() -> Result<(), HError> { +fn main() -> Result<(), HdError> { tracing_subscriber::fmt() .with_env_filter(tracing_subscriber::EnvFilter::from_default_env()) .init(); let addr = "0.0.0.0:21368"; - let server = Arc::new(TcpEventBus::new()); + let server = Arc::new(HdBus::new()); server.publish(EventData::new(1, "Hello"))?; server.publish(EventData::new(1, "World"))?; diff --git a/crates/hd-examples/src/lib.rs b/crates/hd-examples/src/lib.rs new file mode 100644 index 0000000..e69de29 diff --git a/crates/hd-lib/src/bus/mod.rs b/crates/hd-lib/src/bus/mod.rs index a8e70b0..379bb80 100644 --- a/crates/hd-lib/src/bus/mod.rs +++ b/crates/hd-lib/src/bus/mod.rs @@ -1,8 +1,8 @@ pub mod packet; -use tracing::{debug_span, info, info_span}; +use tracing::{debug_span, info}; -use crate::error::HError; +use crate::error::HdError; use std::{ sync::{ Arc, RwLock, @@ -63,7 +63,12 @@ impl EventBusInner { struct SubscriptionHandle { id: SubscriptionId, new_event_signal: Sender, - join_handle: JoinHandle>, + join_handle: JoinHandle>, +} + +pub struct SubscriptionInfo { + pub id: SubscriptionId, + pub event_recv: Receiver>, } pub struct PollResult { @@ -79,8 +84,8 @@ impl EventBus { } } - pub fn publish(&self, evt: EventData) -> Result<(), HError> { - let mut bus = self.inner.write().map_err(|e| HError::BusLockPoisoned(e.to_string()))?; + pub fn publish(&self, evt: EventData) -> Result<(), HdError> { + let mut bus = self.inner.write().map_err(|e| HdError::BusLockPoisoned(e.to_string()))?; let next_event_id = &mut bus.next_event_id; let event = Event { @@ -103,8 +108,8 @@ impl EventBus { Ok(()) } - pub fn get_next_event(&self, cursor: Cursor) -> Result, HError> { - let bus = self.inner.read().map_err(|e| HError::BusLockPoisoned(e.to_string()))?; + pub fn get_next_event(&self, cursor: Cursor) -> Result, HdError> { + let bus = self.inner.read().map_err(|e| HdError::BusLockPoisoned(e.to_string()))?; for evt in &bus.events { if evt.id >= cursor { @@ -112,22 +117,22 @@ impl EventBus { } } - Err(HError::NoMoreEvents) + Err(HdError::NoMoreEvents) } - pub fn subscribe(&self) -> Result>, HError> { - let mut bus = self.inner.write().map_err(|e| HError::BusLockPoisoned(e.to_string()))?; + pub fn subscribe(&self) -> Result { + let mut bus = self.inner.write().map_err(|e| HdError::BusLockPoisoned(e.to_string()))?; let next_subscription_id = bus.next_subscription_id; - // let next_event_id = bus.next_event_id; let next_event_id = 0; let (new_event_signal_sender, new_event_signal_recv) = channel(); let (event_sender, event_recv) = sync_channel(10); + // Start event delivery in the background as and when new events are available let subscription = Subscription::new(next_subscription_id, next_event_id, new_event_signal_recv, event_sender); - let bg_thread_handle = start_delivery(subscription, self.clone()); + // Store a handle that the bus can use signal the bg delivery thread of new events let subscription_handle = SubscriptionHandle { id: next_subscription_id, new_event_signal: new_event_signal_sender, @@ -137,11 +142,17 @@ impl EventBus { bus.next_subscription_id += 1; bus.subscription_handles.push(subscription_handle); - Ok(event_recv) + // Return info to caller that can be used to receive the events and unsubscribe later. + let subscription_info = SubscriptionInfo { + id: next_subscription_id, + event_recv, + }; + + Ok(subscription_info) } - pub fn unsubscribe(&self, id: SubscriptionId) -> Result<(), HError> { - let mut bus = self.inner.write().map_err(|e| HError::BusLockPoisoned(e.to_string()))?; + pub fn unsubscribe(&self, id: SubscriptionId) -> Result<(), HdError> { + let mut bus = self.inner.write().map_err(|e| HdError::BusLockPoisoned(e.to_string()))?; let mut found = None; let mut index = 0; @@ -155,7 +166,7 @@ impl EventBus { } match found { - None => Err(HError::SubscriptionNotFound(format!( + None => Err(HdError::SubscriptionNotFound(format!( "Subscription with id {} is not found", id ))), @@ -193,8 +204,9 @@ impl Subscription { } } -fn start_delivery(mut subscription: Subscription, bus: EventBus) -> JoinHandle> { +fn start_delivery(subscription: Subscription, bus: EventBus) -> JoinHandle> { thread::spawn(move || { + let mut subscription = subscription; loop { let _scope = debug_span!("subscription_delivery", subscription.id).entered(); loop { diff --git a/crates/hd-lib/src/bus/packet.rs b/crates/hd-lib/src/bus/packet.rs index 5f47f7f..ceb02f1 100644 --- a/crates/hd-lib/src/bus/packet.rs +++ b/crates/hd-lib/src/bus/packet.rs @@ -9,25 +9,25 @@ mod packet_constants { #[derive(PartialEq)] pub enum PacketType { Connect, - Subscribe, - Unsubscribe, Publish, - SendEvent, + Deliver, Settle, - Ack, Disconnect, } +#[derive(PartialEq)] +pub enum ConnectMode { + Publish, + Subscribe, +} + impl Display for PacketType { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { let to_write = match self { PacketType::Connect => "CONNECT", - PacketType::Subscribe => "SUBSCRIBE", - PacketType::Unsubscribe => "UNSUBSCRIBE", PacketType::Publish => "PUBLISH", - PacketType::SendEvent => "SEND_EVENT", - PacketType::Settle => "SETTLE_EVENT", - PacketType::Ack => "ACK", + PacketType::Deliver => "DELIVER", + PacketType::Settle => "SETTLE", PacketType::Disconnect => "DISCONNECT", }; @@ -36,18 +36,15 @@ impl Display for PacketType { } pub enum Packet { - Connect, - Subscribe(SubscribePacket), - Unsubscribe, + Connect(ConnectPacket), Publish(PublishPacket), - SendEvent(SendEventPacket), + Deliver(DeliverPacket), Settle(SettlePacket), - Ack(AckPacket), Disconnect, } -pub struct SubscribePacket { - cursor: u64, +pub struct ConnectPacket { + pub mode: ConnectMode, } pub struct PublishPacket { @@ -55,7 +52,7 @@ pub struct PublishPacket { pub data_utf8: Vec, } -pub struct SendEventPacket { +pub struct DeliverPacket { event_id: u64, event_type: u8, data_utf8: Vec, @@ -70,16 +67,8 @@ pub struct AckPacket { } impl Packet { - pub fn create_connect_packet() -> Packet { - Packet::Connect - } - - pub fn create_subscribe_packet(cursor: u64) -> Packet { - Packet::Subscribe(SubscribePacket { cursor }) - } - - pub fn create_unsubscribe_packet() -> Packet { - Packet::Unsubscribe + pub fn create_connect_packet(mode: ConnectMode) -> Packet { + Packet::Connect(ConnectPacket { mode }) } // TODO: Avoid copying data_utf8 @@ -91,8 +80,8 @@ impl Packet { } // TODO: Avoid copying data_utf8 - pub fn create_send_event_packet(evt: Arc) -> Packet { - Packet::SendEvent(SendEventPacket { + pub fn create_deliver_packet(evt: Arc) -> Packet { + Packet::Deliver(DeliverPacket { event_id: evt.id, event_type: evt.data.event_type, data_utf8: evt.data.data_utf8.clone(), @@ -103,10 +92,6 @@ impl Packet { Packet::Settle(SettlePacket { cursor }) } - pub fn create_ack_packet(packet_type: PacketType) -> Packet { - Packet::Ack(AckPacket { packet_type }) - } - pub fn create_disconnect_packet() -> Packet { Packet::Disconnect } @@ -134,7 +119,7 @@ pub struct ReadPacketState { packet_type: PacketType, } -pub fn start_read_packet(buf: &mut [u8; packet_constants::HEADER_SIZE]) -> Result { +pub(crate) fn start_read_packet(buf: &mut [u8; packet_constants::HEADER_SIZE]) -> Result { let packet_type = get_packet_type_from_byte(buf[0])?; let required_buffer_size = read_u16(&buf[1..3])?; @@ -144,14 +129,13 @@ pub fn start_read_packet(buf: &mut [u8; packet_constants::HEADER_SIZE]) -> Resul }) } -pub fn read_packet(buf: &[u8], read_state: ReadPacketState) -> Result { +pub(crate) fn read_packet(buf: &[u8], read_state: ReadPacketState) -> Result { match read_state.packet_type { - PacketType::Connect => Ok(Packet::Connect), - PacketType::Subscribe => { - let cursor = read_u64(&buf[0..])?; - Ok(Packet::Subscribe(SubscribePacket { cursor })) + PacketType::Connect => { + let mode_byte = buf[0]; + let mode = get_connect_mode_from_byte(mode_byte)?; + Ok(Packet::create_connect_packet(mode)) } - PacketType::Unsubscribe => Ok(Packet::Unsubscribe), PacketType::Publish => { let event_type = buf[0]; let event_len = read_u16(&buf[1..3])?; @@ -163,27 +147,23 @@ pub fn read_packet(buf: &[u8], read_state: ReadPacketState) -> Result { + PacketType::Deliver => { let event_id = read_u64(&buf[0..8])?; let event_type = buf[8]; let event_len = read_u16(&buf[9..11])?; - let send_event_packet = SendEventPacket { + let deliver_packet = DeliverPacket { event_id, event_type, data_utf8: buf[11..11 + event_len].to_vec(), }; - Ok(Packet::SendEvent(send_event_packet)) + Ok(Packet::Deliver(deliver_packet)) } PacketType::Settle => { let cursor = read_u64(&buf[0..8])?; - Ok(Packet::Settle(SettlePacket { cursor })) + Ok(Packet::create_settle_packet(cursor)) } - PacketType::Ack => { - let packet_type = get_packet_type_from_byte(buf[0])?; - Ok(Packet::Ack(AckPacket { packet_type })) - } - PacketType::Disconnect => Ok(Packet::Disconnect), + PacketType::Disconnect => Ok(Packet::create_disconnect_packet()), } } @@ -193,7 +173,7 @@ pub struct WritePacketState { packet_type: PacketType, } -pub fn start_write_packet(packet: &Packet) -> WritePacketState { +pub(crate) fn start_write_packet(packet: &Packet) -> WritePacketState { let data_frame_length = calculate_data_frame_length(&packet); let required_buffer_size = packet_constants::HEADER_SIZE + data_frame_length; let packet_type = get_packet_type(packet); @@ -204,7 +184,7 @@ pub fn start_write_packet(packet: &Packet) -> WritePacketState { } } -pub fn write_packet(buf: &mut [u8], packet: &Packet, write_state: WritePacketState) -> Result<(), PacketError> { +pub(crate) fn write_packet(buf: &mut [u8], packet: &Packet, write_state: WritePacketState) -> Result<(), PacketError> { write_header_frame(buf, write_state.packet_type, write_state.data_frame_length); if write_state.data_frame_length > 0 { write_data_frame(&mut buf[3..], packet); @@ -222,13 +202,14 @@ fn write_header_frame(header_frame: &mut [u8], packet_type: PacketType, data_fra fn write_data_frame(data_frame: &mut [u8], packet: &Packet) { match packet { - Packet::Subscribe(connect_packet) => { - write_u64(connect_packet.cursor, data_frame); + Packet::Connect(connect_packet) => { + let connect_mode_byte = get_byte_from_connect_mode(&connect_packet.mode); + data_frame[0] = connect_mode_byte; } Packet::Publish(publish_packet) => { write_event_data(publish_packet.event_type, &publish_packet.data_utf8, data_frame); } - Packet::SendEvent(send_event_packet) => { + Packet::Deliver(send_event_packet) => { write_u64(send_event_packet.event_id, data_frame); write_event_data( send_event_packet.event_type, @@ -239,10 +220,7 @@ fn write_data_frame(data_frame: &mut [u8], packet: &Packet) { Packet::Settle(settle_packet) => { write_u64(settle_packet.cursor, data_frame); } - Packet::Ack(ack_packet) => { - data_frame[0] = get_byte_from_packet_type(&ack_packet.packet_type); - } - _ => {} + Packet::Disconnect => {} } } @@ -287,13 +265,10 @@ fn read_u16(buf: &[u8]) -> Result { pub fn get_packet_type(packet: &Packet) -> PacketType { match packet { - Packet::Connect => PacketType::Connect, - Packet::Subscribe(_) => PacketType::Subscribe, - Packet::Unsubscribe => PacketType::Unsubscribe, + Packet::Connect(_) => PacketType::Connect, Packet::Publish(_) => PacketType::Publish, - Packet::SendEvent(_) => PacketType::SendEvent, + Packet::Deliver(_) => PacketType::Deliver, Packet::Settle(_) => PacketType::Settle, - Packet::Ack(_) => PacketType::Ack, Packet::Disconnect => PacketType::Disconnect, } } @@ -301,45 +276,45 @@ pub fn get_packet_type(packet: &Packet) -> PacketType { fn get_byte_from_packet_type(packet_type: &PacketType) -> u8 { match packet_type { PacketType::Connect => 0, - PacketType::Subscribe => 1, - PacketType::Unsubscribe => 2, - PacketType::Publish => 3, - PacketType::SendEvent => 4, - PacketType::Settle => 5, - PacketType::Ack => 6, - PacketType::Disconnect => 7, + PacketType::Publish => 1, + PacketType::Deliver => 2, + PacketType::Settle => 3, + PacketType::Disconnect => 4, } } fn get_packet_type_from_byte(byte: u8) -> Result { - let packet_type = match byte { - 0 => Some(PacketType::Connect), - 1 => Some(PacketType::Subscribe), - 2 => Some(PacketType::Unsubscribe), - 3 => Some(PacketType::Publish), - 4 => Some(PacketType::SendEvent), - 5 => Some(PacketType::Settle), - 6 => Some(PacketType::Ack), - 7 => Some(PacketType::Disconnect), - _ => None, - }; - - if let Some(packet_type) = packet_type { - return Ok(packet_type); + match byte { + 0 => Ok(PacketType::Connect), + 1 => Ok(PacketType::Publish), + 2 => Ok(PacketType::Deliver), + 3 => Ok(PacketType::Settle), + 4 => Ok(PacketType::Disconnect), + _ => Err(PacketError::WrongPacketType(byte)), } +} - Err(PacketError::WrongPacketType(byte)) +fn get_byte_from_connect_mode(mode: &ConnectMode) -> u8 { + match mode { + ConnectMode::Publish => 0, + ConnectMode::Subscribe => 1, + } +} + +fn get_connect_mode_from_byte(byte: u8) -> Result { + match byte { + 0 => Ok(ConnectMode::Publish), + 1 => Ok(ConnectMode::Subscribe), + _ => Err(PacketError::PacketProtocolError), + } } fn calculate_data_frame_length(packet: &Packet) -> usize { let mut ret = match packet { - Packet::Connect => 0, - Packet::Subscribe(_) => 8, - Packet::Unsubscribe => 0, + Packet::Connect(_) => 1, Packet::Publish(publish_packet) => 1 + 2 + publish_packet.data_utf8.len(), - Packet::SendEvent(send_event_packet) => 8 + 1 + 2 + send_event_packet.data_utf8.len(), + Packet::Deliver(deliver_packet) => 8 + 1 + 2 + deliver_packet.data_utf8.len(), Packet::Settle(_) => 8, - Packet::Ack(_) => 1, Packet::Disconnect => 0, }; diff --git a/crates/hd-lib/src/error/mod.rs b/crates/hd-lib/src/error/mod.rs index 5d42cf1..26ce657 100644 --- a/crates/hd-lib/src/error/mod.rs +++ b/crates/hd-lib/src/error/mod.rs @@ -3,7 +3,7 @@ use std::{fmt::Display, string::FromUtf8Error}; use crate::bus::packet::PacketError; #[derive(Debug)] -pub enum HError { +pub enum HdError { BusLockPoisoned(String), SubscriptionNotFound(String), IoError(std::io::Error), @@ -14,35 +14,35 @@ pub enum HError { PeerDisconnect, } -impl Display for HError { +impl Display for HdError { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { - HError::BusLockPoisoned(s) => write!(f, "{}", s), - HError::SubscriptionNotFound(s) => write!(f, "{}", s), - HError::IoError(error) => write!(f, "{}", error), - HError::NoMoreEvents => write!(f, "{}", "No more events in the bus for now"), - HError::TcpPacketError(packet_error) => write!(f, "Failed to read/write packets due to {}", packet_error), - HError::PeerDisconnect => write!(f, "Peer disconnect"), - HError::Utf8FormatError(e) => write!(f, "Utf8 formatting error {}", e), - HError::ProtocolError => write!(f, "Protocol error"), + HdError::BusLockPoisoned(s) => write!(f, "{}", s), + HdError::SubscriptionNotFound(s) => write!(f, "{}", s), + HdError::IoError(error) => write!(f, "{}", error), + HdError::NoMoreEvents => write!(f, "{}", "No more events in the bus for now"), + HdError::TcpPacketError(packet_error) => write!(f, "Failed to read/write packets due to {}", packet_error), + HdError::PeerDisconnect => write!(f, "Peer disconnect"), + HdError::Utf8FormatError(e) => write!(f, "Utf8 formatting error {}", e), + HdError::ProtocolError => write!(f, "Protocol error"), } } } -impl From for HError { +impl From for HdError { fn from(value: std::io::Error) -> Self { - HError::IoError(value) + HdError::IoError(value) } } -impl From for HError { +impl From for HdError { fn from(value: PacketError) -> Self { - HError::TcpPacketError(value) + HdError::TcpPacketError(value) } } -impl From for HError { +impl From for HdError { fn from(value: FromUtf8Error) -> Self { - HError::Utf8FormatError(value) + HdError::Utf8FormatError(value) } } diff --git a/crates/hd-lib/src/lib.rs b/crates/hd-lib/src/lib.rs index 1662807..a9ebc4d 100644 --- a/crates/hd-lib/src/lib.rs +++ b/crates/hd-lib/src/lib.rs @@ -9,11 +9,11 @@ pub mod hd_tcp { }; use crate::{ - bus::packet::{self, Packet, PacketType}, - error::HError, + bus::packet::{self, Packet}, + error::HdError, }; - pub fn read_packet(stream: &mut TcpStream) -> Result { + pub fn read_packet(stream: &mut TcpStream) -> Result { let mut header_buf = [0u8; 3]; stream.read_exact(&mut header_buf)?; let read_state = packet::start_read_packet(&mut header_buf)?; @@ -22,34 +22,11 @@ pub mod hd_tcp { if read_state.required_buffer_size > 0 { stream.read_exact(&mut buf)?; } - let packet = packet::read_packet(&buf[0..], read_state)?; - if let Packet::Disconnect = packet { - return Err(HError::PeerDisconnect); - }; - - Ok(packet) + Ok(packet::read_packet(&buf[0..], read_state)?) } - pub fn read_ack(stream: &mut TcpStream, packet_types: &[PacketType]) -> Result { - let packet = read_packet(stream)?; - if let Packet::Ack(ack_packet) = &packet { - for packet_type in packet_types.iter() { - if *packet_type == ack_packet.packet_type { - return Ok(packet); - } - } - } - - Err(HError::ProtocolError) - } - - pub fn write_ack(stream: &mut TcpStream, packet_type: PacketType) -> Result<(), HError> { - let packet = Packet::create_ack_packet(packet_type); - write_packet(stream, &packet) - } - - pub fn write_packet(stream: &mut TcpStream, packet: &Packet) -> Result<(), HError> { + pub fn write_packet(stream: &mut TcpStream, packet: &Packet) -> Result<(), HdError> { let write_state = packet::start_write_packet(packet); let mut buf = vec![0u8; write_state.required_buffer_size]; packet::write_packet(&mut buf, packet, write_state)?; diff --git a/crates/hd-lib/src/thread_pool/mod.rs b/crates/hd-lib/src/thread_pool/mod.rs index eb260e3..8b9fb5e 100644 --- a/crates/hd-lib/src/thread_pool/mod.rs +++ b/crates/hd-lib/src/thread_pool/mod.rs @@ -6,9 +6,9 @@ use std::{ thread::{self, JoinHandle}, }; -use tracing::{Level, info, span}; +use tracing::{Level, error, info, span}; -use crate::error::HError; +use crate::error::HdError; pub struct ThreadPool { workers: Vec, @@ -21,7 +21,7 @@ struct Worker { } struct Job { - f: Box Result<(), HError> + Send>, + f: Box Result<(), HdError> + Send>, } impl ThreadPool { @@ -36,9 +36,9 @@ impl ThreadPool { Self { workers, tx } } - pub fn execute(&self, f: F) -> Result<(), HError> + pub fn execute(&self, f: F) -> Result<(), HdError> where - F: FnOnce() -> Result<(), HError>, + F: FnOnce() -> Result<(), HdError>, F: Send + 'static, { let job = Job { f: Box::new(f) }; @@ -54,8 +54,12 @@ impl ThreadPool { loop { let job = rx.lock().unwrap().recv().unwrap(); // Lock is released here info!("Received a new job, executing"); - (job.f)().unwrap(); // TODO: Panic recovery - info!("Finished executing job"); + let result = (job.f)(); + if let Err(e) = result { + error!("Job failed to complete successfully due to error {}", e); + } + + info!("Job execution complete, worker thread is free"); } }); diff --git a/crates/hd-server/Cargo.toml b/crates/hd-server/Cargo.toml index 4fe19da..ed3ca3b 100644 --- a/crates/hd-server/Cargo.toml +++ b/crates/hd-server/Cargo.toml @@ -5,5 +5,4 @@ edition = "2024" [dependencies] tracing = { workspace = true } -tracing-subscriber = { version = "0.3", features = ["env-filter", "fmt"] } -hd-lib = { path = "../hd-lib" } +hd-lib = {path = "../hd-lib"} diff --git a/crates/hd-server/src/connection.rs b/crates/hd-server/src/connection.rs new file mode 100644 index 0000000..e436521 --- /dev/null +++ b/crates/hd-server/src/connection.rs @@ -0,0 +1,185 @@ +use std::{net::TcpStream, time::Duration}; + +use hd_lib::{ + bus::{packet::*, *}, + error::HdError, + hd_tcp, + thread_pool::ThreadPool, +}; +use tracing::{debug, error, info, info_span}; + +#[derive(Copy, Clone, PartialEq)] +enum HdClientConnectionState { + Connect, + Deliver, + Wait, + Disconnect, +} + +struct HdClientConnection { + pub addr: String, + stream: TcpStream, + state: HdClientConnectionState, + subscription: Option, + bus: EventBus, +} + +impl std::fmt::Display for HdClientConnectionState { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + let to_write = match self { + HdClientConnectionState::Connect => "CONNECT", + HdClientConnectionState::Deliver => "DELIVER", + HdClientConnectionState::Wait => "WAIT", + HdClientConnectionState::Disconnect => "DISCONNECT", + }; + + write!(f, "{}", to_write) + } +} + +impl HdClientConnection { + fn new(bus: EventBus, stream: TcpStream) -> Result { + let addr = stream + .peer_addr() + .map(|a| a.to_string()) + .unwrap_or("unknown".to_string()); + + stream.set_read_timeout(Some(Duration::from_secs(5)))?; + + let state = HdClientConnectionState::Connect; + + Ok(Self { + stream, + addr, + state, + bus, + subscription: None, + }) + } +} + +pub(crate) fn handle_client_connection( + thread_pool: &ThreadPool, + bus: EventBus, + stream: TcpStream, +) -> Result<(), HdError> { + thread_pool.execute(move || { + let mut client = HdClientConnection::new(bus, stream)?; + + let _scope = info_span!("handle_client", addr = client.addr,).entered(); + + loop { + if let HdClientConnectionState::Disconnect = client.state { + unsubscribe_client(&mut client)?; + info!("Closing client connection"); + return Ok(()); + } + + let prev_state = client.state; + let next_state = match handle_event_bus_client_state(&mut client) { + Ok(s) => s, + Err(e) => { + error!("Error while handling client {}", e); + HdClientConnectionState::Disconnect + } + }; + + if prev_state != next_state { + info!("State change {} -> {}", prev_state, next_state); + } + + client.state = next_state; + } + }) +} + +fn handle_event_bus_client_state(client: &mut HdClientConnection) -> Result { + match client.state { + HdClientConnectionState::Connect => handle_state_connect(client), + HdClientConnectionState::Deliver => handle_state_deliver_event(client), + HdClientConnectionState::Wait => handle_state_wait_event(client), + HdClientConnectionState::Disconnect => Ok(HdClientConnectionState::Disconnect), + } +} + +fn handle_state_connect(client: &mut HdClientConnection) -> Result { + match hd_tcp::read_packet(&mut client.stream)? { + Packet::Connect(connect_packet) => { + info!("Received connect"); + let next = match connect_packet.mode { + ConnectMode::Publish => HdClientConnectionState::Wait, + ConnectMode::Subscribe => { + let subscription = client.bus.subscribe()?; + client.subscription = Some(subscription); + HdClientConnectionState::Deliver + } + }; + + Ok(next) + } + p => { + error!( + "Received wrong packet of type {}, expected {}. Will disconnect", + get_packet_type(&p), + PacketType::Connect + ); + Ok(HdClientConnectionState::Disconnect) + } + } +} + +fn handle_state_deliver_event(client: &mut HdClientConnection) -> Result { + if let Some(subscription) = &mut client.subscription { + // TODO: When client sends unsubscribe, server shouldn't take forever + // to ack that. + let Ok(evt) = subscription.event_recv.recv() else { + error!("Failed to receive a message for this subscription, disconnecting"); + return Ok(HdClientConnectionState::Disconnect); + }; + + let packet = Packet::create_deliver_packet(evt); + hd_tcp::write_packet(&mut client.stream, &packet)?; + debug!("Published event to client, waiting for ack"); + + let response_packet = hd_tcp::read_packet(&mut client.stream)?; + if let Packet::Disconnect = &response_packet { + info!("Client disconnected"); + // TODO: Allow unsubscribe, and actually disconnect + return Ok(HdClientConnectionState::Disconnect); + } + + debug!("Publish ack received, message is settled"); + } + + Ok(HdClientConnectionState::Deliver) +} + +fn unsubscribe_client(client: &mut HdClientConnection) -> Result<(), HdError> { + if let Some(subscription) = &client.subscription { + client.bus.unsubscribe(subscription.id)?; + client.subscription = None; + }; + + Ok(()) +} + +fn handle_state_wait_event(client: &mut HdClientConnection) -> Result { + match hd_tcp::read_packet(&mut client.stream)? { + Packet::Publish(publish_packet) => { + debug!("Received event from client, publishing to bus"); + let data = String::from_utf8(publish_packet.data_utf8)?; + let evt = EventData::new(publish_packet.event_type, &data); + client.bus.publish(evt)?; + + Ok(HdClientConnectionState::Wait) + } + p => { + error!( + "Received wrong packet of type {}, expected {}. Will disconnect", + get_packet_type(&p), + PacketType::Connect + ); + Ok(HdClientConnectionState::Disconnect) + } + } +} diff --git a/crates/hd-server/src/lib.rs b/crates/hd-server/src/lib.rs index 60ee5f1..d6efb53 100644 --- a/crates/hd-server/src/lib.rs +++ b/crates/hd-server/src/lib.rs @@ -1,29 +1,21 @@ -use std::{ - net::TcpStream, - sync::{Arc, mpsc::Receiver}, - time::Duration, -}; +mod connection; use hd_lib::{ - bus::{ - Event, EventBus, EventData, - packet::{self, Packet, PacketType, PublishPacket, get_packet_type}, - }, - error::HError, - hd_tcp, + bus::{EventBus, EventData}, + error::HdError, thread_pool::ThreadPool, }; -use tracing::{debug, error, info, info_span}; +use tracing::{error, info}; use std::net::TcpListener; -pub struct TcpEventBus { +pub struct HdBus { bus: EventBus, pool: ThreadPool, } -impl TcpEventBus { +impl HdBus { pub fn new() -> Self { let pool = ThreadPool::new(10); Self { @@ -32,18 +24,18 @@ impl TcpEventBus { } } - pub fn publish(&self, evt: EventData) -> Result<(), HError> { + pub fn publish(&self, evt: EventData) -> Result<(), HdError> { self.bus.publish(evt) } } -impl TcpEventBus { - pub fn start(&self, addr: &'static str) -> Result<(), HError> { +impl HdBus { + pub fn start(&self, addr: &'static str) -> Result<(), HdError> { let listener = TcpListener::bind(addr)?; info!("Heimdall running at {}", addr); for stream in listener.incoming() { match stream { - Ok(stream) => handle_event_bus_client(&self.pool, self.bus.clone(), stream)?, + Ok(stream) => connection::handle_client_connection(&self.pool, self.bus.clone(), stream)?, Err(e) => error!("TCP connection failed {}", e), } } @@ -51,199 +43,3 @@ impl TcpEventBus { Ok(()) } } - -#[derive(Copy, Clone, PartialEq)] -enum EventBusClientConnectionState { - Connect, - Command, - SendEvent, - WaitEvent, - Disconnect, -} - -struct EventBusClientConnection { - pub addr: String, - stream: TcpStream, - state: EventBusClientConnectionState, - subscription: Option>>, - 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)))?; - - let state = EventBusClientConnectionState::Connect; - - Ok(Self { - stream, - addr, - state, - bus, - subscription: None, - }) - } -} - -pub fn handle_event_bus_client(thread_pool: &ThreadPool, bus: EventBus, stream: TcpStream) -> Result<(), HError> { - thread_pool.execute(move || { - let mut client = EventBusClientConnection::new(bus, stream)?; - - let _scope = info_span!("handle_client", addr = client.addr,).entered(); - - loop { - if let EventBusClientConnectionState::Disconnect = client.state { - info!("Closing connection due to invalid input"); - return Ok(()); - } - - let prev_state = client.state; - handle_event_bus_client_state(&mut client) - .inspect_err(|e| info!("Closing connection due to error {}", e))?; - if prev_state != client.state { - info!("State change {} -> {}", prev_state, client.state); - } - } - }) -} - -fn handle_event_bus_client_state(client: &mut EventBusClientConnection) -> Result<(), HError> { - let next_state = match client.state { - EventBusClientConnectionState::Connect => handle_state_connect(client), - EventBusClientConnectionState::Command => handle_state_command(client), - EventBusClientConnectionState::SendEvent => handle_state_send_event(client), - EventBusClientConnectionState::WaitEvent => handle_state_wait_event(client), - EventBusClientConnectionState::Disconnect => Ok(EventBusClientConnectionState::Disconnect), - }?; - - client.state = next_state; - Ok(()) -} - -fn handle_state_connect(client: &mut EventBusClientConnection) -> Result { - 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 { - 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 { - if let Some(subscription) = &mut client.subscription { - // TODO: When client sends unsubscribe, server shouldn't take forever - // to ack that. - let Ok(evt) = subscription.recv() else { - error!("Failed to receive a message for this subscription, disconnecting"); - return Ok(EventBusClientConnectionState::Disconnect); - }; - - let packet = Packet::create_send_event_packet(evt); - hd_tcp::write_packet(&mut client.stream, &packet)?; - debug!("Published event to client, waiting for ack"); - - let ack_packet = hd_tcp::read_ack(&mut client.stream, &[PacketType::SendEvent, PacketType::Unsubscribe])?; - if packet::get_packet_type(&ack_packet) == PacketType::Unsubscribe { - info!("Subscriber unsubscribed"); - return Ok(EventBusClientConnectionState::Command); - } - - debug!("Publish ack received, message is settled"); - } - - Ok(EventBusClientConnectionState::SendEvent) -} - -fn handle_state_wait_event(client: &mut EventBusClientConnection) -> Result { - match hd_tcp::read_packet(&mut client.stream)? { - Packet::Publish(publish_packet) => { - debug!("Received event from client, publishing to bus"); - publish_packet_to_bus(client, publish_packet)?; - - hd_tcp::write_ack(&mut client.stream, PacketType::Publish)?; - debug!("Ack receive event complete"); - Ok(EventBusClientConnectionState::WaitEvent) - } - p => { - error!( - "Received wrong packet of type {}, expected {}. Will disconnect", - get_packet_type(&p), - PacketType::Connect - ); - Ok(EventBusClientConnectionState::Disconnect) - } - } -} - -fn publish_packet_to_bus(client: &mut EventBusClientConnection, packet: PublishPacket) -> Result<(), HError> { - let data = String::from_utf8(packet.data_utf8)?; - let evt = EventData::new(packet.event_type, &data); - client.bus.publish(evt)?; - Ok(()) -} - -impl std::fmt::Display for EventBusClientConnectionState { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - let to_write = match self { - EventBusClientConnectionState::Connect => "CONNECT", - EventBusClientConnectionState::Command => "COMMAND", - EventBusClientConnectionState::SendEvent => "SEND_EVENT", - EventBusClientConnectionState::WaitEvent => "WAIT_EVENT", - EventBusClientConnectionState::Disconnect => "DISCONNECT", - }; - - write!(f, "{}", to_write) - } -}