diff --git a/crates/bsnext_client/generated/dto.ts b/crates/bsnext_client/generated/dto.ts index 161f7a9..49d6e3c 100644 --- a/crates/bsnext_client/generated/dto.ts +++ b/crates/bsnext_client/generated/dto.ts @@ -139,7 +139,8 @@ export type InputErrorDTO = | { kind: "MarkdownError", payload: string } | { kind: "Io", payload: string } | { kind: "UnsupportedExtension", payload: string } - | { kind: "MissingExtension", payload: string }; + | { kind: "MissingExtension", payload: string } + | { kind: "EmptyInput", payload: string }; export type ClientEvent = | { kind: "Change", payload: ChangeDTO }; diff --git a/crates/bsnext_client/generated/schema.ts b/crates/bsnext_client/generated/schema.ts index 8031ed7..158f724 100644 --- a/crates/bsnext_client/generated/schema.ts +++ b/crates/bsnext_client/generated/schema.ts @@ -181,6 +181,10 @@ export const inputErrorDTOSchema = z.union([ kind: z.literal("MissingExtension"), payload: z.string(), }), + z.object({ + kind: z.literal("EmptyInput"), + payload: z.string(), + }), ]); export const debounceDTOSchema = z.object({ diff --git a/crates/bsnext_client/inject/dist/index.js b/crates/bsnext_client/inject/dist/index.js index 16ab301..d0f03be 100644 --- a/crates/bsnext_client/inject/dist/index.js +++ b/crates/bsnext_client/inject/dist/index.js @@ -6422,6 +6422,10 @@ var inputErrorDTOSchema = z.union([ z.object({ kind: z.literal("MissingExtension"), payload: z.string() + }), + z.object({ + kind: z.literal("EmptyInput"), + payload: z.string() }) ]); var debounceDTOSchema = z.object({ diff --git a/crates/bsnext_dto/src/lib.rs b/crates/bsnext_dto/src/lib.rs index 6bdb068..6599b62 100644 --- a/crates/bsnext_dto/src/lib.rs +++ b/crates/bsnext_dto/src/lib.rs @@ -292,6 +292,7 @@ pub enum InputErrorDTO { Io(String), UnsupportedExtension(String), MissingExtension(String), + EmptyInput(String), } impl From<&InputError> for InputErrorDTO { @@ -311,6 +312,7 @@ impl From<&InputError> for InputErrorDTO { InputErrorDTO::UnsupportedExtension(e.to_string()) } e @ InputError::MissingExtension(_) => InputErrorDTO::MissingExtension(e.to_string()), + e @ InputError::EmptyInput => InputErrorDTO::EmptyInput(e.to_string()), } } } diff --git a/crates/bsnext_fs/src/inner_fs_event_handler.rs b/crates/bsnext_fs/src/inner_fs_event_handler.rs index a156909..378f26b 100644 --- a/crates/bsnext_fs/src/inner_fs_event_handler.rs +++ b/crates/bsnext_fs/src/inner_fs_event_handler.rs @@ -33,6 +33,7 @@ impl Handler for FsWatcher { path: relative.clone(), }), ctx: self.ctx.clone(), + span: None, }) } } @@ -88,6 +89,7 @@ impl Handler for FsWatcher { recipient.do_send(FsEvent { kind: evt, ctx: self.ctx.clone(), + span: None, }) } } diff --git a/crates/bsnext_fs/src/lib.rs b/crates/bsnext_fs/src/lib.rs index e286630..def545d 100644 --- a/crates/bsnext_fs/src/lib.rs +++ b/crates/bsnext_fs/src/lib.rs @@ -11,7 +11,9 @@ pub mod watch_path_handler; mod watcher; use std::path::{Path, PathBuf}; +use std::sync::Arc; use std::time::Duration; +use tracing::Span; // use tokio_stream::StreamExt; @@ -77,6 +79,7 @@ impl Default for FsEventContext { pub struct FsEvent { pub kind: FsEventKind, pub ctx: FsEventContext, + pub span: Option>, } #[derive(Debug, Clone)] @@ -84,7 +87,8 @@ pub enum FsEventKind { Change(ChangeEvent), ChangeBuffered(BufferedChangeEvent), PathAdded(PathAddedEvent), - PathRemoved(PathRemovedEvent), + PathRemoved(PathEvent), + PathNotFoundError(PathEvent), } #[derive(actix::Message, Debug, Clone)] @@ -130,7 +134,7 @@ pub struct PathAddedEvent { #[derive(actix::Message, Debug, Clone)] #[rtype(result = "()")] -pub struct PathRemovedEvent { +pub struct PathEvent { pub path: PathBuf, } diff --git a/crates/bsnext_fs/src/remove_path_handler.rs b/crates/bsnext_fs/src/remove_path_handler.rs index e7bebf3..6c748db 100644 --- a/crates/bsnext_fs/src/remove_path_handler.rs +++ b/crates/bsnext_fs/src/remove_path_handler.rs @@ -1,5 +1,5 @@ use crate::actor::FsWatcher; -use crate::{FsEvent, FsEventKind, FsWatchError, PathRemovedEvent}; +use crate::{FsEvent, FsEventKind, FsWatchError, PathEvent}; use actix::Handler; use notify::Watcher; use std::path::PathBuf; @@ -27,12 +27,13 @@ impl Handler for FsWatcher { } }; for recip in &self.receivers { - let evt = FsEventKind::PathRemoved(PathRemovedEvent { + let evt = FsEventKind::PathRemoved(PathEvent { path: relative.clone(), }); recip.do_send(FsEvent { kind: evt, ctx: self.ctx.clone(), + span: None, }) } } diff --git a/crates/bsnext_fs/src/stop_handler.rs b/crates/bsnext_fs/src/stop_handler.rs index fd7e8e4..cdd4e75 100644 --- a/crates/bsnext_fs/src/stop_handler.rs +++ b/crates/bsnext_fs/src/stop_handler.rs @@ -1,14 +1,17 @@ use crate::actor::FsWatcher; use actix::{ActorContext, Handler}; +use std::sync::Arc; +use tracing::{instrument, Span}; #[derive(actix::Message, Debug, Clone)] #[rtype(result = "()")] -pub struct StopWatcher; +pub struct StopWatcher(pub Arc); impl Handler for FsWatcher { type Result = (); - fn handle(&mut self, _msg: StopWatcher, ctx: &mut Self::Context) -> Self::Result { + #[instrument(skip_all, name = "StopWatcher for FsWatcher", parent=msg.0.id())] + fn handle(&mut self, msg: StopWatcher, ctx: &mut Self::Context) -> Self::Result { self.watcher = None; ctx.stop(); } diff --git a/crates/bsnext_fs/src/test/mod.rs b/crates/bsnext_fs/src/test/mod.rs index f0641ec..81b12a3 100644 --- a/crates/bsnext_fs/src/test/mod.rs +++ b/crates/bsnext_fs/src/test/mod.rs @@ -5,11 +5,13 @@ use actix::{Actor, Addr}; use std::fs::File; use std::io::Write; use std::path::{Path, PathBuf}; +use std::sync::Arc; use std::time::Duration; use tempfile::TempDir; use crate::filter::Filter; use tokio::time::sleep; +use tracing::Span; struct A { events: Vec, @@ -76,6 +78,7 @@ impl TestCase { let r = RequestWatchPath { recipients: vec![self.recip_addr.clone().recipient()], path: self.dir.to_path_buf(), + span: Arc::new(Span::none()), }; let _ = self.addr.send(r).await; @@ -121,6 +124,7 @@ impl TestCase { .collect(), FsEventKind::PathAdded(_) => vec![], FsEventKind::PathRemoved(_) => vec![], + FsEventKind::PathNotFoundError(_) => vec![], }) .map(|pb| pb.file_name().unwrap().to_string_lossy().to_string()) .collect() diff --git a/crates/bsnext_fs/src/watch_path_handler.rs b/crates/bsnext_fs/src/watch_path_handler.rs index 4de110a..26e7116 100644 --- a/crates/bsnext_fs/src/watch_path_handler.rs +++ b/crates/bsnext_fs/src/watch_path_handler.rs @@ -1,23 +1,29 @@ use crate::actor::FsWatcher; -use crate::{FsEvent, FsEventKind, FsWatchError, PathAddedEvent}; +use crate::{FsEvent, FsEventKind, PathAddedEvent, PathEvent}; use actix::{ActorContext, Handler, Recipient}; use notify::{RecursiveMode, Watcher}; use std::path::PathBuf; +use std::sync::Arc; +use tracing::{trace_span, Span}; #[derive(actix::Message)] -#[rtype(result = "Result<(), FsWatchError>")] +#[rtype(result = "()")] pub struct RequestWatchPath { pub recipients: Vec>, pub path: PathBuf, + pub span: Arc, } impl Handler for FsWatcher { - type Result = Result<(), FsWatchError>; + type Result = (); // todo: ensure this isn't sent for every input change - #[tracing::instrument(skip_all, name = "RequestWatchPath for FsWatcher")] fn handle(&mut self, msg: RequestWatchPath, _ctx: &mut Self::Context) -> Self::Result { - tracing::trace!(path = ?msg.path, "-> WatchPath"); + let span = trace_span!(parent: msg.span.id(), "RequestWatchPath for FsWatcher", ?msg.path); + let s = Arc::new(span); + let span_c = s.clone(); + let _guard = s.enter(); + // tracing::trace!(path = ?msg.path, "-> WatchPath"); if let Some(watcher) = self.watcher.as_mut() { match watcher.watch(&msg.path, RecursiveMode::Recursive) { Ok(_) => { @@ -53,16 +59,25 @@ impl Handler for FsWatcher { recip.do_send(FsEvent { kind: evt, ctx: self.ctx.clone(), + span: Some(span_c.clone()), }) } } Err(err) => { tracing::error!("cannot watch: {}", err); + for recip in &msg.recipients { + let evt = FsEventKind::PathNotFoundError(PathEvent { + path: msg.path.clone(), + }); + recip.do_send(FsEvent { + kind: evt, + ctx: self.ctx.clone(), + span: Some(span_c.clone()), + }) + } _ctx.stop(); - return Err(FsWatchError::Watcher(err)); } } } - Ok(()) } } diff --git a/crates/bsnext_input/src/lib.rs b/crates/bsnext_input/src/lib.rs index 7a6c644..5b88bac 100644 --- a/crates/bsnext_input/src/lib.rs +++ b/crates/bsnext_input/src/lib.rs @@ -39,6 +39,11 @@ impl Input { } fn from_yaml_path>(path: P) -> Result { let str = read_to_string(&path)?; + if str.trim().is_empty() { + return Err(InputError::YamlError(YamlError::EmptyError { + path: path.as_ref().to_string_lossy().to_string(), + })); + } let output = serde_yaml::from_str::(str.as_str()).map_err(move |e| { if let Some(location) = e.location() { YamlError::ParseErrorWithLocation { @@ -71,6 +76,8 @@ impl Input { pub enum InputError { #[error("no suitable inputs could be found")] MissingInputs, + #[error("input file is empty")] + EmptyInput, #[error("could not read input, error: {0}")] InvalidInput(String), #[error("io error")] @@ -132,10 +139,10 @@ pub enum DirError { } #[derive(Debug, thiserror::Error, serde::Serialize, serde::Deserialize)] -pub struct PathDefs(Vec); +pub struct PathDefs(pub Vec); #[derive(Debug, thiserror::Error, serde::Serialize, serde::Deserialize)] -struct PathDefinition { +pub struct PathDefinition { pub input: String, pub cwd: PathBuf, pub absolute: PathBuf, diff --git a/crates/bsnext_input/src/yml.rs b/crates/bsnext_input/src/yml.rs index f08307b..431acdc 100644 --- a/crates/bsnext_input/src/yml.rs +++ b/crates/bsnext_input/src/yml.rs @@ -39,4 +39,6 @@ original error: input: String, serde_error: serde_yaml::Error, }, + #[error("Input file was empty: {path}")] + EmptyError { path: String }, } diff --git a/crates/bsnext_output/src/pretty.rs b/crates/bsnext_output/src/pretty.rs index f342d91..d8e2ab4 100644 --- a/crates/bsnext_output/src/pretty.rs +++ b/crates/bsnext_output/src/pretty.rs @@ -43,7 +43,7 @@ impl OutputWriter for PrettyPrint { ) -> anyhow::Result<()> { match evt { StartupEvent::Started => { - write!(sink, "{}", Line::prefixed().info("started..."))?; + writeln!(sink, "{}", Line::prefixed().info("started..."))?; } StartupEvent::FailedStartup(err) => { writeln!( @@ -196,6 +196,7 @@ fn print_input_error( InputErrorDTO::Io(evt) => evt, InputErrorDTO::UnsupportedExtension(evt) => evt, InputErrorDTO::MissingExtension(evt) => evt, + InputErrorDTO::EmptyInput(evt) => evt, }; writeln!(w, "{}", Line::unprefixed().indent(indent).error(v))?; Ok(()) diff --git a/crates/bsnext_output/src/tests.rs b/crates/bsnext_output/src/tests.rs index 517603b..47984b7 100644 --- a/crates/bsnext_output/src/tests.rs +++ b/crates/bsnext_output/src/tests.rs @@ -1,10 +1,10 @@ use crate::pretty::PrettyPrint; use crate::OutputWriter; use bsnext_dto::{ - ExternalEvents, GetServersMessageResponse, IdentityDTO, ServerChange, - ServerChangeSet, ServerChangeSetItem, ServerDTO, ServersStarted, StartupEvent, + ExternalEvents, GetServersMessageResponse, IdentityDTO, ServerChange, ServerChangeSet, + ServerChangeSetItem, ServerDTO, ServersStarted, StartupEvent, }; -use std::io::{BufWriter}; +use std::io::BufWriter; fn iden_1() -> IdentityDTO { IdentityDTO::Address { diff --git a/crates/bsnext_system/src/cli.rs b/crates/bsnext_system/src/cli.rs index bfb62fb..c622e00 100644 --- a/crates/bsnext_system/src/cli.rs +++ b/crates/bsnext_system/src/cli.rs @@ -66,7 +66,17 @@ where let stdout = &mut std::io::stdout(); match startup_oneshot_receiver.await? { - Ok(DidStart::Started) => tracing::info!("started..."), + Ok(DidStart::Started) => { + let evt = StartupEvent::Started; + match printer.handle_startup_event(stdout, &evt) { + Ok(_) => {} + Err(e) => tracing::error!(?e), + }; + match stdout.flush() { + Ok(_) => {} + Err(e) => tracing::error!("could not flush {e}"), + }; + } Err(e) => { let evt = StartupEvent::FailedStartup((&e).into()); match printer.handle_startup_event(stdout, &evt) { diff --git a/crates/bsnext_system/src/lib.rs b/crates/bsnext_system/src/lib.rs index 21edb67..62ffa28 100644 --- a/crates/bsnext_system/src/lib.rs +++ b/crates/bsnext_system/src/lib.rs @@ -79,7 +79,6 @@ impl BsSystem { let s = Arc::new(span); let c = s.clone(); let _c2 = s.clone(); - let _g = s.enter(); let route_watchables = to_route_watchables(input); let server_watchables = to_server_watchables(input); @@ -169,7 +168,10 @@ impl BsSystem { #[tracing::instrument(skip(self))] fn publish_external_event(&mut self, evt: ExternalEvents) { - tracing::debug!(?evt); + match evt { + ExternalEvents::InputError(_) => tracing::error!(?evt), + _ => tracing::debug!(?evt), + } let outgoing = EventWithSpan { evt }; if let Some(external_event_sender) = &self.external_event_sender { Arbiter::current().spawn({ diff --git a/crates/bsnext_system/src/monitor.rs b/crates/bsnext_system/src/monitor.rs index 0387297..2386423 100644 --- a/crates/bsnext_system/src/monitor.rs +++ b/crates/bsnext_system/src/monitor.rs @@ -5,18 +5,22 @@ use std::hash::Hash; use bsnext_core::servers_supervisor::file_changed_handler::{FileChanged, FilesChanged}; use bsnext_dto::{ExternalEvents, StoppedWatching, Watching}; use bsnext_fs::watch_path_handler::RequestWatchPath; -use bsnext_fs::{Debounce, FsEvent, FsEventContext, FsEventKind}; +use bsnext_fs::{ + BufferedChangeEvent, ChangeEvent, Debounce, FsEvent, FsEventContext, FsEventKind, + PathAddedEvent, PathEvent, +}; use bsnext_input::route::{ DebounceDuration, DirRoute, FilterKind, RouteKind, Spec, SpecOpts, WatchOpts, }; use bsnext_input::server_config::Identity; -use bsnext_input::Input; +use bsnext_input::{Input, InputError, PathDefinition, PathDefs, PathError}; use std::path::{Path, PathBuf}; +use std::sync::Arc; use std::time::Duration; use bsnext_fs::actor::FsWatcher; -use tracing::{span, trace_span, Level}; +use tracing::trace_span; #[derive(Debug, Clone)] pub struct Monitor { @@ -35,6 +39,10 @@ impl actix::Handler for BsSystem { type Result = (); fn handle(&mut self, msg: MonitorInput, ctx: &mut Self::Context) -> Self::Result { + let span = trace_span!("monitor_input", ?msg.path, ?msg.cwd); + let s = Arc::new(span); + let span_c = s.clone(); + let _guard = s.enter(); let mut input_watcher = bsnext_fs::actor::FsWatcher::for_input(&msg.cwd, 0); // todo: does this need to be configurable (eg: by main config)? @@ -50,98 +58,121 @@ impl actix::Handler for BsSystem { input_watcher_addr.do_send(RequestWatchPath { recipients: vec![ctx.address().recipient()], path: msg.path.to_path_buf(), + span: span_c.clone(), }); } } -impl actix::Handler for BsSystem { - type Result = (); +impl BsSystem { + #[tracing::instrument(skip(self))] + fn handle_buffered( + &mut self, + msg: &FsEvent, + buf: &BufferedChangeEvent, + ) -> Option { + tracing::debug!(msg.event_count = buf.events.len(), msg.ctx = ?msg.ctx, ?buf); + let paths = buf + .events + .iter() + .map(|evt| evt.absolute.to_owned()) + .collect::>(); + let as_strings = paths + .iter() + .map(|p| p.to_string_lossy().to_string()) + .collect::>(); + if let Some(servers) = &self.servers_addr { + servers.do_send(FilesChanged { + paths: paths.clone(), + id: msg.ctx.id(), + }) + } + // todo(alpha): need to exclude changes to the input file if this event has captured it + Some(ExternalEvents::FilesChanged(bsnext_dto::FilesChangedDTO { + paths: as_strings, + })) + } + fn handle_change(&mut self, msg: &FsEvent, inner: &ChangeEvent) -> Option { + let span = trace_span!("handle_change", ?inner.absolute_path); + let _guard = span.enter(); + match msg.ctx { + FsEventContext::InputFile { id: _ } => { + tracing::info!(?inner, "InputFile file changed"); + let input = Input::from_input_path(&inner.absolute_path); - fn handle(&mut self, msg: FsEvent, _ctx: &mut Self::Context) -> Self::Result { - match msg.kind { - FsEventKind::ChangeBuffered(buffer_change) => { - let span = span!(Level::TRACE, "FsEventKind::ChangeBuffered"); - let _guard = span.enter(); - tracing::debug!(msg.event_count = buffer_change.events.len(), msg.ctx = ?msg.ctx, ?buffer_change); - // let id = msg.ctx_id(); - let paths = buffer_change - .events - .iter() - .map(|evt| evt.absolute.to_owned()) - .collect::>(); - let as_strings = paths - .iter() - .map(|p| p.to_string_lossy().to_string()) - .collect::>(); + let Ok(input) = input else { + let err = input.unwrap_err(); + return Some(ExternalEvents::InputError(err.into())); + }; + + self.accept_input(&input); + self.inform_servers(input); + + Some(ExternalEvents::InputFileChanged( + bsnext_dto::FileChanged::from_path_buf(&inner.path), + )) + } + FsEventContext::Other { id } => { + tracing::trace!(?inner, "Other file changed"); + // todo: tie these changed to an input identity? if let Some(servers) = &self.servers_addr { - servers.do_send(FilesChanged { - paths: paths.clone(), - id: msg.ctx.id(), + servers.do_send(FileChanged { + path: inner.absolute_path.clone(), + id, }) } - // todo(alpha): need to exclude changes to the input file if this event has captured it - let evt = - ExternalEvents::FilesChanged(bsnext_dto::FilesChangedDTO { paths: as_strings }); - self.publish_external_event(evt); - } - FsEventKind::Change(inner) => { - let span = trace_span!("FsEventKind::Change", ?inner.absolute_path); - let _guard = span.enter(); - match msg.ctx { - FsEventContext::InputFile { id: _ } => { - tracing::info!(?inner, "InputFile file changed"); - let input = Input::from_input_path(&inner.absolute_path); - - let Ok(input) = input else { - tracing::debug!("ignoring FsWatchEvent because the input was invalid"); - let err = input.unwrap_err(); - tracing::error!(?err, "{}", err); - return; - }; - - tracing::debug!("InputFile file was deserialized"); - - self.accept_input(&input); - self.inform_servers(input); - - let evt = ExternalEvents::InputFileChanged( - bsnext_dto::FileChanged::from_path_buf(&inner.path), - ); - self.publish_external_event(evt); - } - FsEventContext::Other { id } => { - tracing::trace!(?inner, "Other file changed"); - // todo: tie these changed to an input identity? - if let Some(servers) = &self.servers_addr { - servers.do_send(FileChanged { - path: inner.absolute_path.clone(), - id, - }) - } - let evt = ExternalEvents::FileChanged( - bsnext_dto::FileChanged::from_path_buf(&inner.path), - ); - - self.publish_external_event(evt); - } - } - } - FsEventKind::PathAdded(path) => { - let span = trace_span!("FsEventKind::PathAdded", ?path); - let _guard = span.enter(); - let evt = - ExternalEvents::Watching(Watching::from_path_buf(&path.path, path.debounce)); - self.publish_external_event(evt); - } - FsEventKind::PathRemoved(path) => { - let span = trace_span!("FsEventKind::PathRemoved", ?path); - let _guard = span.enter(); - let evt = - ExternalEvents::WatchingStopped(StoppedWatching::from_path_buf(&path.path)); - self.publish_external_event(evt); + Some(ExternalEvents::FileChanged( + bsnext_dto::FileChanged::from_path_buf(&inner.path), + )) } } } + #[tracing::instrument(skip(self))] + fn handle_path_added(&mut self, path: &PathAddedEvent) -> Option { + Some(ExternalEvents::Watching(Watching::from_path_buf( + &path.path, + path.debounce, + ))) + } + + #[tracing::instrument(skip(self))] + fn handle_path_removed(&mut self, path: &PathEvent) -> Option { + Some(ExternalEvents::WatchingStopped( + StoppedWatching::from_path_buf(&path.path), + )) + } + + #[tracing::instrument(skip(self))] + fn handle_path_not_found(&mut self, pdo: &PathEvent) -> Option { + let as_str = pdo.path.to_string_lossy().to_string(); + let cwd = self.cwd.clone().unwrap(); + let abs = cwd.join(&as_str); + let def = PathDefinition { + input: as_str, + cwd: self.cwd.clone().unwrap(), + absolute: abs, + }; + let e = InputError::PathError(PathError::MissingPaths { + paths: PathDefs(vec![def]), + }); + Some(ExternalEvents::InputError(e.into())) + } +} + +impl actix::Handler for BsSystem { + type Result = (); + #[tracing::instrument(skip(self, _ctx), name = "FsEvent handler for BsSystem", parent=msg.span.as_ref().and_then(|s|s.id()))] + fn handle(&mut self, msg: FsEvent, _ctx: &mut Self::Context) -> Self::Result { + let next = match &msg.kind { + FsEventKind::ChangeBuffered(buffer_change) => self.handle_buffered(&msg, buffer_change), + FsEventKind::Change(inner) => self.handle_change(&msg, inner), + FsEventKind::PathAdded(path) => self.handle_path_added(path), + FsEventKind::PathRemoved(path) => self.handle_path_removed(path), + FsEventKind::PathNotFoundError(pdo) => self.handle_path_not_found(pdo), + }; + if let Some(ext) = next { + self.publish_external_event(ext) + } + } } #[derive(Debug, PartialEq, PartialOrd, Ord, Eq, Hash, Clone)] diff --git a/crates/bsnext_system/src/monitor_any_watchables.rs b/crates/bsnext_system/src/monitor_any_watchables.rs index 61f21d0..24fc8c1 100644 --- a/crates/bsnext_system/src/monitor_any_watchables.rs +++ b/crates/bsnext_system/src/monitor_any_watchables.rs @@ -11,7 +11,7 @@ use std::collections::BTreeSet; use std::path::PathBuf; use std::sync::Arc; use std::time::Duration; -use tracing::{debug_span, Span}; +use tracing::Span; #[derive(actix::Message)] #[rtype(result = "()")] @@ -24,9 +24,12 @@ pub struct MonitorAnyWatchables { impl actix::Handler for BsSystem { type Result = (); + #[tracing::instrument(skip_all, name="BsSystem handler for MonitorAnyWatchables", parent=msg.span.id())] fn handle(&mut self, msg: MonitorAnyWatchables, ctx: &mut Self::Context) -> Self::Result { - let s = debug_span!(parent: msg.span.id(), "BsSystem handler for MonitorAnyWatchables"); - let _g = s.enter(); + let s = Span::current(); + let span = Arc::new(s); + let span_c = span.clone(); + let _g = span.enter(); tracing::debug!("MonitorAnyWatchables {:?}", msg.watchables); tracing::trace!("MonitorAnyWatchables {:#?}", msg.watchables); @@ -42,7 +45,7 @@ impl actix::Handler for BsSystem { for any_watchable in to_remove { if let Some(mon) = self.any_monitors.get(any_watchable) { - mon.addr.do_send(StopWatcher); + mon.addr.do_send(StopWatcher(span.clone())); ctx.notify(DropMonitor((*any_watchable).clone())) } } @@ -87,6 +90,7 @@ impl actix::Handler for BsSystem { monitor.addr.do_send(RequestWatchPath { recipients: vec![ctx.address().recipient()], path: monitor.path.clone(), + span: span_c.clone(), }); ctx.notify(InsertMonitor((*watchable).clone(), monitor))