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

rskit_server/
component.rs
