diff --git a/Cargo.lock b/Cargo.lock index 8831ed0..e38b20f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2,10 +2,238 @@ # It is not intended for manual editing. version = 4 +[[package]] +name = "aho-corasick" +version = "1.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c982642fa9e8606056828ee9a8505737230110bb1099153c79efe865c59d12ba" +dependencies = [ + "memchr", +] + +[[package]] +name = "cfg-if" +version = "1.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" + [[package]] name = "hd-bus" version = "0.1.0" +dependencies = [ + "tracing", + "tracing-subscriber", +] [[package]] name = "hd-client" version = "0.1.0" +dependencies = [ + "tracing", +] + +[[package]] +name = "lazy_static" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe" + +[[package]] +name = "log" +version = "0.4.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f9f8bd3e56ce4dfc153cf470fffbfa98c7620958b312ca5c3a4b8d5181fd13c6" + +[[package]] +name = "matchers" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d1525a2a28c7f4fa0fc98bb91ae755d1e2d1505079e05539e35bc876b5d65ae9" +dependencies = [ + "regex-automata", +] + +[[package]] +name = "memchr" +version = "2.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98" + +[[package]] +name = "nu-ansi-term" +version = "0.50.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" +dependencies = [ + "windows-sys", +] + +[[package]] +name = "once_cell" +version = "1.21.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" + +[[package]] +name = "pin-project-lite" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" + +[[package]] +name = "proc-macro2" +version = "1.0.107" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "985e7ec9bb745e6ce6535b544d84d6cd6f7ad8bd711c398938ae983b91a766d9" +dependencies = [ + "unicode-ident", +] + +[[package]] +name = "quote" +version = "1.0.47" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1fbf4db142a473a8d80c26bbf18454ed458bf8d26c8219c331daecfdbd079001" +dependencies = [ + "proc-macro2", +] + +[[package]] +name = "regex-automata" +version = "0.4.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ad8553b9b26413251cbf30e620595c7a41b3887f03da04579c0e6b0d6a06b4b2" +dependencies = [ + "aho-corasick", + "memchr", + "regex-syntax", +] + +[[package]] +name = "regex-syntax" +version = "0.8.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d6f6ff9a378485b298a5286656da665ba74413d36db0979633275d2e708145d4" + +[[package]] +name = "sharded-slab" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f40ca3c46823713e0d4209592e8d6e826aa57e928f09752619fc696c499637f6" +dependencies = [ + "lazy_static", +] + +[[package]] +name = "smallvec" +version = "1.15.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8ed6a63f02c8539c91a8685a86f4099661ba3da017932f6ebbea6de3f0fa7c90" + +[[package]] +name = "syn" +version = "2.0.119" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "872831b642d1a07999a962a351ed35b955ea2cfc8f3862091e2a240a84f17297" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + +[[package]] +name = "thread_local" +version = "1.1.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1ad99c4c6d32803332c548b1af0540b357b3f5fc0be8f6c6bfe8b2e6ae784070" +dependencies = [ + "cfg-if", +] + +[[package]] +name = "tracing" +version = "0.1.44" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "63e71662fa4b2a2c3a26f570f037eb95bb1f85397f3cd8076caed2f026a6d100" +dependencies = [ + "pin-project-lite", + "tracing-attributes", + "tracing-core", +] + +[[package]] +name = "tracing-attributes" +version = "0.1.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7490cfa5ec963746568740651ac6781f701c9c5ea257c58e057f3ba8cf69e8da" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "tracing-core" +version = "0.1.36" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "db97caf9d906fbde555dd62fa95ddba9eecfd14cb388e4f491a66d74cd5fb79a" +dependencies = [ + "once_cell", + "valuable", +] + +[[package]] +name = "tracing-log" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ee855f1f400bd0e5c02d150ae5de3840039a3f54b025156404e34c23c03f47c3" +dependencies = [ + "log", + "once_cell", + "tracing-core", +] + +[[package]] +name = "tracing-subscriber" +version = "0.3.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cb7f578e5945fb242538965c2d0b04418d38ec25c79d160cd279bf0731c8d319" +dependencies = [ + "matchers", + "nu-ansi-term", + "once_cell", + "regex-automata", + "sharded-slab", + "smallvec", + "thread_local", + "tracing", + "tracing-core", + "tracing-log", +] + +[[package]] +name = "unicode-ident" +version = "1.0.24" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75" + +[[package]] +name = "valuable" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65" + +[[package]] +name = "windows-link" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" + +[[package]] +name = "windows-sys" +version = "0.61.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc" +dependencies = [ + "windows-link", +] diff --git a/Cargo.toml b/Cargo.toml index 7bef8be..d5639e4 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -3,3 +3,6 @@ members = [ "crates/*" ] resolver = "2" + +[workspace.dependencies] +tracing = "0.1.44" diff --git a/crates/hd-bus/Cargo.toml b/crates/hd-bus/Cargo.toml index 5596574..5d2e459 100644 --- a/crates/hd-bus/Cargo.toml +++ b/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"] } diff --git a/crates/hd-bus/src/bin/server.rs b/crates/hd-bus/src/bin/server.rs index f6ff119..95a04d0 100644 --- a/crates/hd-bus/src/bin/server.rs +++ b/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(()) } diff --git a/crates/hd-bus/src/bus.rs b/crates/hd-bus/src/bus.rs index f8c36e8..4c71863 100644 --- a/crates/hd-bus/src/bus.rs +++ b/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> { 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(()); } diff --git a/crates/hd-bus/src/connection.rs b/crates/hd-bus/src/connection.rs index 2008853..8f89e7d 100644 --- a/crates/hd-bus/src/connection.rs +++ b/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 ID: {} TYPE: {} LEN: {} <=", evt.id, evt.data.event_type, data_len); let stream = &mut client.stream; stream.write(&mut buf).map_err(HError::TcpPeerError)?; diff --git a/crates/hd-bus/src/lib.rs b/crates/hd-bus/src/lib.rs index 6ce72d6..8be3f68 100644 --- a/crates/hd-bus/src/lib.rs +++ b/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), } } diff --git a/crates/hd-bus/src/thread_pool.rs b/crates/hd-bus/src/thread_pool.rs index 0fd6105..85df88c 100644 --- a/crates/hd-bus/src/thread_pool.rs +++ b/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>>) -> 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"); } }); diff --git a/crates/hd-client/Cargo.toml b/crates/hd-client/Cargo.toml index 3622bd1..d3862b7 100644 --- a/crates/hd-client/Cargo.toml +++ b/crates/hd-client/Cargo.toml @@ -4,3 +4,4 @@ version = "0.1.0" edition = "2024" [dependencies] +tracing = { workspace = true }