diff options
Diffstat (limited to '')
| -rw-r--r-- | crates/daemon/src/daemon.rs (renamed from crates/turtle/src/atuin_daemon/daemon.rs) | 190 |
1 files changed, 22 insertions, 168 deletions
diff --git a/crates/turtle/src/atuin_daemon/daemon.rs b/crates/daemon/src/daemon.rs index 80aaeef8..70e65c1e 100644 --- a/crates/turtle/src/atuin_daemon/daemon.rs +++ b/crates/daemon/src/daemon.rs @@ -4,20 +4,19 @@ //! //! - [`DaemonState`]: Shared state owned by the daemon //! - [`DaemonHandle`]: A lightweight, cloneable handle for accessing daemon state -//! - [`Component`]: A trait for implementing daemon components //! - [`Daemon`]: The main daemon orchestrator //! - [`DaemonBuilder`]: Builder for constructing and configuring the daemon use std::sync::Arc; -use crate::atuin_client::{ +use crate::aclient::{ database::ClientSqlite as HistoryDatabase, encryption, record::sqlite_store::SqliteStore, settings::Settings, }; use eyre::{Context, Result}; use tokio::sync::{RwLock, broadcast}; -use crate::atuin_daemon::events::DaemonEvent; +use crate::events::DaemonEvent; // ============================================================================ // DaemonState @@ -25,7 +24,7 @@ use crate::atuin_daemon::events::DaemonEvent; /// Shared state owned by the daemon. /// -/// This contains all the resources that components and services need access to. +/// This contains all the resources that services need access to. /// The state is wrapped in an `Arc` and accessed via [`DaemonHandle`]. pub(crate) struct DaemonState { // Event bus @@ -48,7 +47,7 @@ pub(crate) struct DaemonState { /// A lightweight handle to the daemon's shared state. /// -/// This is the primary way for components, gRPC services, and spawned tasks to +/// This is the primary way for gRPC services, and spawned tasks to /// interact with the daemon. It provides access to: /// /// - Event emission and subscription @@ -88,12 +87,24 @@ impl DaemonHandle { tracing::warn!("failed to emit event (no receivers?): {e}"); } } + pub(crate) async fn wait_for(&self, matches: fn(&DaemonEvent) -> bool) -> Result<DaemonEvent> { + let mut rx = self.subscribe(); + loop { + match rx.recv().await { + Ok(e) if matches(&e) => { + return Ok(e); + } + Err(err) => { + return Err(err).context("while waiting for events"); + } + Ok(_) => (), + } + } + } /// Subscribe to the event bus. /// /// Returns a receiver that will receive all events emitted after this call. - /// Useful for components that need to listen for events outside of the - /// normal `handle_event` callback flow. pub(crate) fn subscribe(&self) -> broadcast::Receiver<DaemonEvent> { self.state.event_tx.subscribe() } @@ -113,16 +124,6 @@ impl DaemonHandle { self.state.settings.read().await } - /// Apply already-loaded settings and emit a [`SettingsReloaded`] event. - /// - /// Use this when settings have already been loaded (e.g., from a file watcher) - /// to avoid parsing the config file twice. - pub(crate) async fn apply_settings(&self, settings: Settings) { - *self.state.settings.write().await = settings; - self.emit(DaemonEvent::SettingsReloaded); - tracing::info!("settings applied"); - } - /// Get the encryption key. pub(crate) fn encryption_key(&self) -> &[u8; 32] { &self.state.encryption_key @@ -148,92 +149,12 @@ impl std::fmt::Debug for DaemonHandle { } // ============================================================================ -// Component Trait -// ============================================================================ - -/// A daemon component that handles a specific domain. -/// -/// Components are the building blocks of the daemon. Each component: -/// -/// - Has a unique name for logging and debugging -/// - Can optionally expose gRPC services -/// - Receives a [`DaemonHandle`] on startup for accessing daemon resources -/// - Handles events from the event bus -/// - Performs cleanup on shutdown -/// -/// # Lifecycle -/// -/// 1. **Construction**: Component is created (usually via `new()`) -/// 2. **Start**: `start()` is called with a [`DaemonHandle`] -/// 3. **Running**: `handle_event()` is called for each event on the bus -/// 4. **Shutdown**: `stop()` is called for cleanup -/// -/// # Example -/// -/// ```ignore -/// pub(crate) struct MyComponent { -/// handle: Option<DaemonHandle>, -/// } -/// -/// #[async_trait] -/// impl Component for MyComponent { -/// fn name(&self) -> &'static str { "my-component" } -/// -/// async fn start(&mut self, handle: DaemonHandle) -> Result<()> { -/// self.handle = Some(handle); -/// Ok(()) -/// } -/// -/// async fn handle_event(&mut self, event: &DaemonEvent) -> Result<()> { -/// match event { -/// DaemonEvent::SomeEvent => { -/// // Handle the event -/// if let Some(handle) = &self.handle { -/// handle.emit(DaemonEvent::ResponseEvent); -/// } -/// } -/// _ => {} -/// } -/// Ok(()) -/// } -/// -/// async fn stop(&mut self) -> Result<()> { -/// Ok(()) -/// } -/// } -/// ``` -#[tonic::async_trait] -pub(crate) trait Component: Send + Sync { - /// Human-readable name for logging and debugging. - fn name(&self) -> &'static str; - - /// Called once at startup. - /// - /// Store the handle if you need to emit events or access daemon resources - /// later. The handle is cheaply cloneable, so feel free to clone it for - /// spawned tasks. - async fn start(&mut self, handle: DaemonHandle) -> Result<()>; - - /// Handle an incoming event. - /// - /// Called for every event on the bus. To emit new events in response, - /// use the handle stored during `start()`. Events emitted here will be - /// processed in subsequent event loop iterations. - async fn handle_event(&mut self, event: &DaemonEvent) -> Result<()>; - - /// Called on graceful shutdown. - /// - /// Use this to clean up resources, abort spawned tasks, etc. - async fn stop(&mut self) -> Result<()>; -} - -// ============================================================================ // Daemon // ============================================================================ /// The main daemon orchestrator. /// -/// The daemon manages components, runs the event loop, and coordinates startup +/// The daemon runs the event loop, and coordinates startup /// and shutdown. It is constructed via [`DaemonBuilder`]. /// /// # Event Loop @@ -241,14 +162,11 @@ pub(crate) trait Component: Send + Sync { /// The daemon runs a simple event loop: /// /// 1. Wait for an event on the bus -/// 2. Dispatch the event to all components (in registration order) -/// 3. Components may emit new events in response /// 4. Repeat until `ShutdownRequested` is received /// /// Events emitted during handling are queued and processed in subsequent /// iterations, ensuring the loop eventually drains. pub(crate) struct Daemon { - components: Vec<Box<dyn Component>>, handle: DaemonHandle, } @@ -265,26 +183,8 @@ impl Daemon { self.handle.clone() } - /// Start all components. - /// - /// This must be called before `run_event_loop()`. It initializes all - /// registered components with the daemon handle. - pub(crate) async fn start_components(&mut self) -> Result<()> { - for component in &mut self.components { - tracing::info!(component = component.name(), "starting component"); - component - .start(self.handle.clone()) - .await - .with_context(|| format!("failed to start component: {}", component.name()))?; - } - Ok(()) - } - /// Run the daemon event loop. - /// - /// This processes events until a [`ShutdownRequested`] event is received. - /// Components must be started first via `start_components()`. - pub(crate) async fn run_event_loop(&mut self) -> Result<()> { + pub(crate) async fn wait_for_shutdown(&mut self) -> Result<()> { let mut event_rx = self.handle.subscribe(); loop { match event_rx.recv().await { @@ -293,8 +193,7 @@ impl Daemon { break; } Ok(event) => { - tracing::debug!(?event, "processing event"); - self.dispatch_event(&event).await; + tracing::debug!(?event, "event received"); } Err(broadcast::error::RecvError::Lagged(n)) => { tracing::warn!( @@ -310,35 +209,6 @@ impl Daemon { } Ok(()) } - - /// Stop all components. - /// - /// This performs graceful shutdown of all components. - pub(crate) async fn stop_components(&mut self) { - for component in &mut self.components { - tracing::info!(component = component.name(), "stopping component"); - if let Err(e) = component.stop().await { - tracing::error!( - component = component.name(), - error = ?e, - "error stopping component" - ); - } - } - tracing::info!("all components stopped"); - } - - async fn dispatch_event(&mut self, event: &DaemonEvent) { - for component in &mut self.components { - if let Err(e) = component.handle_event(event).await { - tracing::error!( - component = component.name(), - error = ?e, - "error handling event" - ); - } - } - } } // ============================================================================ @@ -353,9 +223,6 @@ impl Daemon { /// let daemon = Daemon::builder(settings) /// .store(store) /// .history_db(history_db) -/// .component(HistoryComponent::new()) -/// .component(SearchComponent::new()) -/// .component(SyncComponent::new()) /// .build() /// .await?; /// @@ -365,7 +232,6 @@ pub(crate) struct DaemonBuilder { settings: Settings, store: Option<SqliteStore>, history_db: Option<HistoryDatabase>, - components: Vec<Box<dyn Component>>, } impl DaemonBuilder { @@ -375,7 +241,6 @@ impl DaemonBuilder { settings, store: None, history_db: None, - components: Vec::new(), } } @@ -391,14 +256,6 @@ impl DaemonBuilder { self } - /// Register a component. - /// - /// Components are started in registration order and stopped in reverse order. - pub(crate) fn component(mut self, component: impl Component + 'static) -> Self { - self.components.push(Box::new(component)); - self - } - /// Build the daemon. /// /// This loads the encryption key and creates the daemon state. @@ -428,9 +285,6 @@ impl DaemonBuilder { // Create the handle (just a reference to the state) let handle = DaemonHandle { state }; - Ok(Daemon { - components: self.components, - handle, - }) + Ok(Daemon { handle }) } } |