WIP

crates/hd-bus/Cargo.toml

@@ -4,3 +4,5 @@ version = "0.1.0"
 edition = "2024"
 
 [dependencies]
+tracing = { workspace = true }
+tracing-subscriber = { version = "0.3", features = ["env-filter", "fmt"] }

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

@@ -1,10 +1,26 @@
+use std::{sync::Arc, thread, time::Duration};
+
 use hd_bus::{TcpEventBus, bus::EventData, error::HError};
+use tracing::info_span;
 
 fn main() -> Result<(), HError> {
-    let server = TcpEventBus::new();
+    tracing_subscriber::fmt()
+        .with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
+        .init();
+
+    let addr = "0.0.0.0:21368";
+    let _scope = info_span!("example_server").entered();
+    let server = Arc::new(TcpEventBus::new());
     server.publish(EventData::new(1, "Hello"))?;
     server.publish(EventData::new(1, "World"))?;
 
-    let addr = "0.0.0.0:21368";
-    server.start(addr)
+    let server_clone = server.clone();
+    thread::spawn(move || server.start(addr));
+
+    for i in 0..255 {
+        thread::sleep(Duration::from_secs(5));
+        server_clone.publish(EventData::new(i, "asd"))?;
+    }
+
+    Ok(())
 }

crates/hd-bus/src/bus.rs

@@ -1,3 +1,5 @@
+use tracing::{info, info_span};
+
 use crate::error::HError;
 use std::{
     sync::{
@@ -91,7 +93,7 @@ impl EventBus {
 
         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);
+                tracing::error!("Failed to notify {}", e);
             }
         }
 
@@ -191,11 +193,12 @@ impl Subscription {
 fn start_delivery(mut subscription: Subscription, bus: EventBus) -> JoinHandle<Result<(), HError>> {
     thread::spawn(move || {
         loop {
+            let _scope = info_span!("subscription_delivery", subscription.id).entered();
             loop {
                 if let Ok(evt) = bus.get_next_event(subscription.cursor) {
-                    eprintln!("Delivering event {} to subscription {}", evt.id, subscription.id);
+                    info!("Delivering event {} ", evt.id);
                     if let Err(e) = subscription.client.send(evt) {
-                        eprintln!("Subscription delivery for id {} failed {}", subscription.id, e);
+                        tracing::error!("Event delivery for id {} failed {}", subscription.id, e);
                         return Ok(());
                     }
 

crates/hd-bus/src/connection.rs

@@ -5,6 +5,8 @@ use std::{
     time::Duration,
 };
 
+use tracing::{Span, info, info_span};
+
 use crate::{
     bus::{Event, EventBus},
     error::HError,
@@ -56,18 +58,27 @@ pub fn handle_event_bus_client(thread_pool: &ThreadPool, bus: EventBus, stream:
     thread_pool.execute(move || {
         let mut client = EventBusClientConnection::new(bus, stream)?;
 
+        let _scope = info_span!(
+            "handle_event_bus_client",
+            addr = client.addr,
+            state = %client.state
+        )
+        .entered();
+
         loop {
             if let EventBusClientConnectionState::Disconnect = client.state {
-                eprintln!("[{}] Closing connection due to invalid input", client.addr);
+                info!("Closing connection due to invalid input");
                 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))?;
+                .inspect_err(|e| info!("Closing connection due to error {}", e))?;
             if prev_state != client.state {
-                println!("[{}] State change {} -> {}", client.addr, prev_state, client.state);
+                info!("State change {} -> {}", prev_state, client.state);
             }
+
+            Span::current().record("state", format!("{}", client.state));
         }
     })
 }
@@ -118,11 +129,7 @@ fn handle_state_send_event(client: &mut EventBusClientConnection) -> Result<Even
                 buf[i] = evt.data.data_utf8[i - 17];
             }
 
-            print!("Sending message : ");
-            for i in &buf {
-                print!("{} ", i);
-            }
-            println!("End");
+            info!("SEND => ID: {} TYPE: {} LEN: {} <=", evt.id, evt.data.event_type, data_len);
 
             let stream = &mut client.stream;
             stream.write(&mut buf).map_err(HError::TcpPeerError)?;

crates/hd-bus/src/lib.rs

@@ -3,6 +3,8 @@ mod connection;
 pub mod error;
 mod thread_pool;
 
+use tracing::{Level, event, info, span};
+
 use crate::{
     bus::{EventBus, EventData},
     connection::handle_event_bus_client,
@@ -33,11 +35,11 @@ impl TcpEventBus {
 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);
+        info!("Heimdall listening 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),
+                Err(e) => tracing::error!("TCP connection failed {}", e),
             }
         }
 

crates/hd-bus/src/thread_pool.rs

@@ -6,6 +6,8 @@ use std::{
     thread::{self, JoinHandle},
 };
 
+use tracing::{Level, event, info, span};
+
 use crate::error::HError;
 
 pub struct ThreadPool {
@@ -47,11 +49,13 @@ impl ThreadPool {
 
     fn worker_thread(id: usize, rx: Arc<Mutex<Receiver<Job>>>) -> Worker {
         let handle = thread::spawn(move || {
+            let s = span!(Level::INFO, "worker_thread", id = 1);
+            let _scope = s.enter();
             loop {
                 let job = rx.lock().unwrap().recv().unwrap(); // Lock is released here
-                println!("Worker [{}] received a new job, executing", id);
+                info!("Received a new job, executing");
                 (job.f)().unwrap(); // TODO: Panic recovery
-                println!("Worker [{}] finished executing job", id);
+                info!("Finished executing job");
             }
         });
 

crates/hd-client/Cargo.toml

@@ -4,3 +4,4 @@ version = "0.1.0"
 edition = "2024"
 
 [dependencies]
+tracing = { workspace = true }