app_forge_kit_service/
observer.rs

1use crate::Error;
2use crate::signal;
3use app_forge_kit_telemetry_tracing::debug;
4use async_trait::async_trait;
5use std::cell::RefCell;
6use std::sync::Arc;
7use std::sync::atomic::{AtomicUsize, Ordering};
8use tokio::task::JoinSet;
9use tokio::time;
10
11#[async_trait]
12pub trait Observable: Send + Sync {
13    fn init(&self) -> Result<(), Error> {
14        Ok(())
15    }
16    async fn serve(&self) -> Result<(), Error> {
17        Ok(())
18    }
19    async fn signal(&self, _signal: &signal::Signal) -> Result<(), Error> {
20        Ok(())
21    }
22}
23
24#[derive(Default, PartialEq, Eq)]
25enum ObservableState {
26    #[default]
27    Registered,
28    Initialized,
29    Started,
30}
31
32pub struct Resource {
33    observable: Arc<Box<dyn Observable>>,
34    state: RefCell<ObservableState>,
35}
36
37impl Resource {
38    pub fn new(observable: Box<dyn Observable>) -> Self {
39        Resource {
40            observable: Arc::new(observable),
41            state: RefCell::new(ObservableState::default()),
42        }
43    }
44}
45
46pub struct Observer {
47    resources: RefCell<Vec<Resource>>,
48    signals: RefCell<Vec<signal::Signal>>,
49}
50
51const OBSERVER_RESOURCES_CAPACITY: usize = 8;
52const OBSERVER_SIGNALS_CAPACITY: usize = 4;
53
54const SIGNAL_LOOP_SLEEP_DURATION_MICROS: u64 = 10;
55
56#[allow(clippy::new_without_default)]
57impl Observer {
58    pub fn new() -> Self {
59        let instance = Self {
60            resources: RefCell::new(Vec::with_capacity(OBSERVER_RESOURCES_CAPACITY)),
61            signals: RefCell::new(Vec::with_capacity(OBSERVER_SIGNALS_CAPACITY)),
62        };
63
64        instance
65            .signals
66            .borrow_mut()
67            .extend([signal::Signal::SIGINT, signal::Signal::SIGTERM]);
68
69        instance
70    }
71
72    pub fn register(&self, resource: Box<dyn Observable>) -> Result<(), Error> {
73        self.resources.borrow_mut().push(Resource::new(resource));
74
75        Ok(())
76    }
77
78    fn init(&self) -> Result<(), Error> {
79        debug!("initializing");
80
81        self.resources
82            .borrow()
83            .iter()
84            .try_for_each(|resource| -> Result<(), Error> {
85                resource.observable.init()?;
86                resource.state.replace(ObservableState::Initialized);
87
88                Ok(())
89            })
90    }
91
92    fn register_signals_listeners(
93        signals: &[signal::Signal],
94        signal_catcher: &Arc<AtomicUsize>,
95    ) -> Result<(), Error> {
96        for signal in signals {
97            signal_hook::flag::register_usize(
98                signal.into(),
99                signal_catcher.clone(),
100                signal.into(),
101            )?;
102        }
103
104        Ok(())
105    }
106
107    async fn signal_catcher(
108        signals: &[signal::Signal],
109        observables: &[Arc<Box<dyn Observable>>],
110    ) -> Result<(), Error> {
111        let signal_catcher = Arc::new(AtomicUsize::new(signal::Signal::UNKNOWN.into()));
112
113        Self::register_signals_listeners(signals, &signal_catcher)?;
114
115        let mut sleep_duration = time::Duration::from_micros(0);
116
117        loop {
118            if !sleep_duration.is_zero() {
119                time::sleep(sleep_duration).await;
120            }
121
122            let signal_value: signal::Signal = signal_catcher.load(Ordering::Acquire).into();
123
124            if signal_value.is_unknown() {
125                if sleep_duration.is_zero() {
126                    sleep_duration = time::Duration::from_micros(SIGNAL_LOOP_SLEEP_DURATION_MICROS);
127                }
128
129                continue;
130            }
131
132            debug!("catching signal: {}", signal_value);
133
134            for observable in observables.iter() {
135                observable.signal(&signal_value).await?;
136            }
137
138            if signal_value.is_terminate() {
139                debug!("terminating");
140
141                break;
142            }
143        }
144
145        Ok(())
146    }
147
148    fn spawn_signal_catcher(&self, join_set: &mut JoinSet<Result<(), Error>>) -> Result<(), Error> {
149        let signals: Arc<Vec<signal::Signal>> =
150            Arc::new(self.signals.borrow().iter().cloned().collect());
151
152        let observables: Arc<Vec<Arc<Box<dyn Observable>>>> = Arc::new(
153            self.resources
154                .borrow()
155                .iter()
156                .map(|resource| resource.observable.clone())
157                .collect(),
158        );
159
160        join_set.spawn(async move { Self::signal_catcher(&signals, &observables).await });
161
162        Ok(())
163    }
164
165    async fn join_handle(join_set: &mut JoinSet<Result<(), Error>>) -> Result<(), Error> {
166        let mut result: Result<(), Error> = Ok(());
167
168        while let Some(join_handle) = join_set.join_next().await {
169            let mut need_shutdown = false;
170
171            match join_handle {
172                Ok(join_handle_result) => {
173                    if let Err(err) = join_handle_result {
174                        result = Err(err);
175                        need_shutdown = true;
176                    }
177                }
178                Err(err) => {
179                    if err.is_panic() {
180                        result = Err(err.into());
181                    }
182                    need_shutdown = true;
183                }
184            }
185
186            if need_shutdown {
187                join_set.join_next().await;
188                break;
189            }
190        }
191
192        result
193    }
194
195    pub async fn run(&self) -> Result<(), Error> {
196        self.init()?;
197
198        let mut join_set: JoinSet<Result<(), Error>> = JoinSet::new();
199
200        debug!("starting");
201        self.spawn_signal_catcher(&mut join_set)?;
202
203        for resource in self.resources.borrow().iter() {
204            let observable = Arc::clone(&resource.observable);
205
206            join_set.spawn(async move { observable.serve().await });
207
208            resource.state.replace(ObservableState::Started);
209        }
210
211        Self::join_handle(&mut join_set).await
212    }
213}
app_forge_kit_service/observer.rs

app_forge_kit_service/
observer.rs
