rskit_server/
component.rs

1use std::net::SocketAddr;
2use std::sync::Arc;
3use std::time::Duration;
4
5use parking_lot::Mutex;
6use tokio_util::sync::CancellationToken;
7
8use rskit_bootstrap::{Component, Health};
9use rskit_errors::{AppError, AppResult};
10
11use crate::config::GrpcServerConfig;
12
13// ---------------------------------------------------------------------------
14// GrpcServer
15// ---------------------------------------------------------------------------
16
17/// A running gRPC server managed as a bootstrap [`Component`].
18///
19/// `GrpcServerBuilder::build()` produces this; it is then registered with
20/// [`rskit_bootstrap::Registry`] so the app lifecycle starts and stops it.
21pub struct GrpcServer {
22    name: String,
23    config: GrpcServerConfig,
24    start_fn:
25        Arc<dyn Fn(SocketAddr, CancellationToken) -> tokio::task::JoinHandle<()> + Send + Sync>,
26    cancel: CancellationToken,
27    handle: Mutex<Option<tokio::task::JoinHandle<()>>>,
28}
29
30impl GrpcServer {
31    pub(crate) fn new(
32        name: String,
33        config: GrpcServerConfig,
34        start_fn: Arc<
35            dyn Fn(SocketAddr, CancellationToken) -> tokio::task::JoinHandle<()> + Send + Sync,
36        >,
37    ) -> Self {
38        Self {
39            name,
40            config,
41            start_fn,
42            cancel: CancellationToken::new(),
43            handle: Mutex::new(None),
44        }
45    }
46}
47
48#[async_trait::async_trait]
49impl Component for GrpcServer {
50    fn name(&self) -> &str {
51        &self.name
52    }
53
54    async fn start(&self) -> AppResult<()> {
55        let addr: SocketAddr =
56            self.config
57                .addr()
58                .parse()
59                .map_err(|e: std::net::AddrParseError| {
60                    AppError::new(
61                        rskit_errors::ErrorCode::Internal,
62                        format!("invalid gRPC address '{}': {}", self.config.addr(), e),
63                    )
64                })?;
65
66        tracing::info!(component = %self.name, addr = %addr, "starting gRPC server");
67
68        let cancel = self.cancel.clone();
69        let jh = (self.start_fn)(addr, cancel);
70
71        *self.handle.lock() = Some(jh);
72
73        Ok(())
74    }
75
76    async fn stop(&self) -> AppResult<()> {
77        tracing::info!(component = %self.name, "stopping gRPC server");
78        self.cancel.cancel();
79
80        let handle = self.handle.lock().take();
81        if let Some(jh) = handle {
82            let _ = tokio::time::timeout(Duration::from_secs(10), jh).await;
83        }
84
85        Ok(())
86    }
87
88    fn health(&self) -> Health {
89        let running = self
90            .handle
91            .lock()
92            .as_ref()
93            .map(|h| !h.is_finished())
94            .unwrap_or(false);
95        if running {
96            Health::healthy(&self.name)
97        } else {
98            Health::unhealthy(&self.name, "server not running")
99        }
100    }
101}
rskit_server/component.rs

rskit_server/
component.rs
