chips/heimdall
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactors

Browse Source
This commit is contained in:
cool-mist
2026-10-03 15:56:29 +05:30
parent a3385e7702
commit c9438d00a6
14 changed files with 358 additions and 401 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/hd-lib/src/bus/mod.rs
+29 -17
View File
@@ -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<u64>,
join_handle: JoinHandle<Result<(), HError>>,
join_handle: JoinHandle<Result<(), HdError>>,
}
pub struct SubscriptionInfo {
pub id: SubscriptionId,
pub event_recv: Receiver<Arc<Event>>,
}
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<Arc<Event>, HError> {
let bus = self.inner.read().map_err(|e| HError::BusLockPoisoned(e.to_string()))?;
pub fn get_next_event(&self, cursor: Cursor) -> Result<Arc<Event>, 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<Receiver<Arc<Event>>, HError> {
let mut bus = self.inner.write().map_err(|e| HError::BusLockPoisoned(e.to_string()))?;
pub fn subscribe(&self) -> Result<SubscriptionInfo, HdError> {
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<Result<(), HError>> {
fn start_delivery(subscription: Subscription, bus: EventBus) -> JoinHandle<Result<(), HdError>> {
thread::spawn(move || {
let mut subscription = subscription;
loop {
let _scope = debug_span!("subscription_delivery", subscription.id).entered();
loop {
crates/hd-lib/src/bus/packet.rs
+65 -90
View File
@@ -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<u8>,
}
pub struct SendEventPacket {
pub struct DeliverPacket {
event_id: u64,
event_type: u8,
data_utf8: Vec<u8>,
@@ -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<Event>) -> Packet {
Packet::SendEvent(SendEventPacket {
pub fn create_deliver_packet(evt: Arc<Event>) -> 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<ReadPacketState, PacketError> {
pub(crate) fn start_read_packet(buf: &mut [u8; packet_constants::HEADER_SIZE]) -> Result<ReadPacketState, PacketError> {
let packet_type = get_packet_type_from_byte(buf[0])?;
let required_buffer_size = read_u16(&buf[1..3])?;
@@ -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<Packet, PacketError> {
pub(crate) fn read_packet(buf: &[u8], read_state: ReadPacketState) -> Result<Packet, PacketError> {
match read_state.packet_type {
PacketType::Connect => Ok(Packet::Connect),
PacketType::Subscribe => {
let cursor = read_u64(&buf[0..])?;
Ok(Packet::Subscribe(SubscribePacket { cursor }))
PacketType::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<Packet, Pa
};
Ok(Packet::Publish(publish_packet))
}
PacketType::SendEvent => {
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<usize, PacketError> {
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<PacketType, PacketError> {
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<ConnectMode, PacketError> {
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,
};
crates/hd-lib/src/error/mod.rs
+16 -16
View File
@@ -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<std::io::Error> for HError {
impl From<std::io::Error> for HdError {
fn from(value: std::io::Error) -> Self {
HError::IoError(value)
HdError::IoError(value)
}
}
impl From<PacketError> for HError {
impl From<PacketError> for HdError {
fn from(value: PacketError) -> Self {
HError::TcpPacketError(value)
HdError::TcpPacketError(value)
}
}
impl From<FromUtf8Error> for HError {
impl From<FromUtf8Error> for HdError {
fn from(value: FromUtf8Error) -> Self {
HError::Utf8FormatError(value)
HdError::Utf8FormatError(value)
}
}
crates/hd-lib/src/lib.rs
+5 -28
View File
@@ -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<Packet, HError> {
pub fn read_packet(stream: &mut TcpStream) -> Result<Packet, HdError> {
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<Packet, HError> {
let packet = read_packet(stream)?;
if let Packet::Ack(ack_packet) = &packet {
for packet_type in packet_types.iter() {
if *packet_type == ack_packet.packet_type {
return Ok(packet);
}
}
}
Err(HError::ProtocolError)
}
pub fn write_ack(stream: &mut TcpStream, packet_type: PacketType) -> Result<(), HError> {
let packet = Packet::create_ack_packet(packet_type);
write_packet(stream, &packet)
}
pub fn write_packet(stream: &mut TcpStream, packet: &Packet) -> Result<(), HError> {
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)?;
crates/hd-lib/src/thread_pool/mod.rs
+11 -7
View File
@@ -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<Worker>,
@@ -21,7 +21,7 @@ struct Worker {
}
struct Job {
f: Box<dyn FnOnce() -> Result<(), HError> + Send>,
f: Box<dyn FnOnce() -> Result<(), HdError> + Send>,
}
impl ThreadPool {
@@ -36,9 +36,9 @@ impl ThreadPool {
Self { workers, tx }
}
pub fn execute<F>(&self, f: F) -> Result<(), HError>
pub fn execute<F>(&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");
}
});