#![allow(clippy::pedantic)]
use std::any::Any;
use std::error::Error as StdError;
use std::fmt;
use std::sync::Arc;
use async_trait::async_trait;
use schemars::JsonSchema;
use serde::de::DeserializeOwned;
use super::lifecycle::ShutdownToken;
use behest_core::health::HealthStatus;
#[derive(Clone)]
pub struct ComponentContext {
shutdown: ShutdownToken,
}
impl ComponentContext {
#[must_use]
pub fn new(shutdown: ShutdownToken) -> Self {
Self { shutdown }
}
#[must_use]
pub fn shutdown(&self) -> ShutdownToken {
self.shutdown.clone()
}
#[must_use]
pub fn child_shutdown(&self) -> ShutdownToken {
self.shutdown.child()
}
}
impl fmt::Debug for ComponentContext {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("ComponentContext").finish_non_exhaustive()
}
}
#[async_trait]
pub trait Component: Send + Sync + 'static {
const NAME: &'static str;
type Config: DeserializeOwned + JsonSchema + Send + Sync + 'static;
type Error: StdError + Send + Sync + 'static;
async fn init(cfg: &Self::Config, ctx: &ComponentContext) -> Result<Self, Self::Error>
where
Self: Sized;
async fn start(&self) -> Result<(), Self::Error> {
Ok(())
}
async fn stop(&self) -> Result<(), Self::Error> {
Ok(())
}
async fn health(&self) -> HealthStatus {
HealthStatus::healthy()
}
fn depends_on() -> &'static [&'static str] {
&[]
}
async fn pre_replace_hook(&self) -> Result<(), Self::Error> {
Ok(())
}
async fn post_replace_hook(&self) -> Result<(), Self::Error> {
Ok(())
}
}
pub trait AnyComponent: Send + Sync + 'static {
fn name(&self) -> &'static str;
fn as_any_arc(&self) -> Arc<dyn Any + Send + Sync>;
fn start(&self) -> futures_util::future::BoxFuture<'_, Result<(), AnyComponentError>>;
fn stop(&self) -> futures_util::future::BoxFuture<'_, Result<(), AnyComponentError>>;
fn health(&self) -> futures_util::future::BoxFuture<'_, HealthStatus>;
fn pre_replace(&self) -> futures_util::future::BoxFuture<'_, Result<(), AnyComponentError>>;
fn post_replace(&self) -> futures_util::future::BoxFuture<'_, Result<(), AnyComponentError>>;
}
#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum AnyComponentError {
#[error("component `{name}` failed: {message}")]
Component {
name: String,
message: String,
},
#[error("component `{0}` is not initialized")]
NotInitialized(String),
}
#[cfg(test)]
mod tests {
use super::*;
use async_trait::async_trait;
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};
#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
struct DummyConfig {
label: String,
}
struct DummyComponent {
label: String,
}
#[async_trait]
impl Component for DummyComponent {
const NAME: &'static str = "test.dummy";
type Config = DummyConfig;
type Error = std::io::Error;
async fn init(cfg: &Self::Config, _ctx: &ComponentContext) -> Result<Self, Self::Error> {
Ok(Self {
label: cfg.label.clone(),
})
}
}
#[tokio::test]
async fn component_init_constructs_with_config() {
let shutdown = ShutdownToken::new();
let ctx = ComponentContext::new(shutdown);
let cfg = DummyConfig {
label: "alpha".into(),
};
let c = DummyComponent::init(&cfg, &ctx).await.unwrap_or_else(|e| {
panic!("init failed: {e}");
});
assert_eq!(c.label, "alpha");
}
#[tokio::test]
async fn default_lifecycle_is_noop() {
let shutdown = ShutdownToken::new();
let ctx = ComponentContext::new(shutdown);
let c = DummyComponent::init(&DummyConfig { label: "x".into() }, &ctx)
.await
.unwrap_or_else(|e| panic!("{e}"));
let _ = c.start().await;
let _ = c.stop().await;
assert!(c.health().await.is_healthy());
}
}