a3s_boot/app/
microservice.rs

1#[cfg(feature = "shutdown-hooks")]
2use super::shutdown::{normalize_shutdown_signals, wait_for_shutdown_signal, ShutdownSignal};
3use super::BootApplication;
4use crate::{MessageTransport, Result};
5#[cfg(feature = "shutdown-hooks")]
6use std::future::Future;
7
8/// Managed standalone microservice built from Boot message patterns.
9pub struct BootMicroservice<T> {
10    app: BootApplication,
11    transport: T,
12    initialized: bool,
13    #[cfg(feature = "shutdown-hooks")]
14    shutdown_signals: Option<Vec<ShutdownSignal>>,
15}
16
17impl<T> BootMicroservice<T>
18where
19    T: MessageTransport,
20{
21    pub fn new(app: BootApplication, transport: T) -> Self {
22        Self {
23            app,
24            transport,
25            initialized: false,
26            #[cfg(feature = "shutdown-hooks")]
27            shutdown_signals: None,
28        }
29    }
30
31    pub fn app(&self) -> &BootApplication {
32        &self.app
33    }
34
35    pub fn transport(&self) -> &T {
36        &self.transport
37    }
38
39    pub fn into_app(self) -> BootApplication {
40        self.app
41    }
42
43    pub fn build_client(&self) -> Result<T::Output> {
44        self.transport.build(self.app.clone())
45    }
46
47    pub fn is_initialized(&self) -> bool {
48        self.initialized
49    }
50
51    #[cfg(feature = "shutdown-hooks")]
52    pub fn enable_shutdown_hooks<I>(&mut self, signals: I) -> &mut Self
53    where
54        I: IntoIterator<Item = ShutdownSignal>,
55    {
56        self.shutdown_signals = Some(normalize_shutdown_signals(signals));
57        self
58    }
59
60    #[cfg(feature = "shutdown-hooks")]
61    pub fn enable_default_shutdown_hooks(&mut self) -> &mut Self {
62        self.enable_shutdown_hooks(ShutdownSignal::default_signals())
63    }
64
65    pub async fn init(&mut self) -> Result<()> {
66        if self.initialized {
67            return Ok(());
68        }
69
70        if let Err(error) = self.app.bootstrap().await {
71            let _ = self.app.shutdown().await;
72            return Err(error);
73        }
74
75        self.initialized = true;
76        Ok(())
77    }
78
79    pub async fn close(&mut self) -> Result<()> {
80        self.close_inner(None).await
81    }
82
83    pub async fn close_with_signal(&mut self, signal: impl Into<String>) -> Result<()> {
84        self.close_inner(Some(signal.into())).await
85    }
86
87    async fn close_inner(&mut self, signal: Option<String>) -> Result<()> {
88        if !self.initialized {
89            return Ok(());
90        }
91
92        match signal {
93            Some(signal) => self.app.shutdown_with_signal(signal).await?,
94            None => self.app.shutdown().await?,
95        }
96        self.initialized = false;
97        Ok(())
98    }
99
100    pub async fn listen(&mut self) -> Result<()> {
101        #[cfg(feature = "shutdown-hooks")]
102        if let Some(signals) = self.shutdown_signals.clone() {
103            return self
104                .listen_with_shutdown_signal_future(wait_for_shutdown_signal(signals))
105                .await;
106        }
107
108        self.init().await?;
109        let serve_result = self.transport.serve(self.app.clone()).await;
110        let close_result = self.close().await;
111
112        match (serve_result, close_result) {
113            (Err(error), _) => Err(error),
114            (Ok(()), Err(error)) => Err(error),
115            (Ok(()), Ok(())) => Ok(()),
116        }
117    }
118
119    #[cfg(feature = "shutdown-hooks")]
120    pub(crate) async fn listen_with_shutdown_signal_future<F>(
121        &mut self,
122        signal_future: F,
123    ) -> Result<()>
124    where
125        F: Future<Output = Result<ShutdownSignal>> + Send,
126    {
127        self.init().await?;
128        let serve_future = self.transport.serve(self.app.clone());
129        futures_util::pin_mut!(signal_future);
130        futures_util::pin_mut!(serve_future);
131
132        tokio::select! {
133            serve_result = &mut serve_future => {
134                let close_result = self.close().await;
135                match (serve_result, close_result) {
136                    (Err(error), _) => Err(error),
137                    (Ok(()), Err(error)) => Err(error),
138                    (Ok(()), Ok(())) => Ok(()),
139                }
140            }
141            signal_result = &mut signal_future => {
142                match signal_result {
143                    Ok(signal) => self.close_with_signal(signal.as_str()).await,
144                    Err(error) => {
145                        let _ = self.close().await;
146                        Err(error)
147                    }
148                }
149            }
150        }
151    }
152}
a3s_boot/app/microservice.rs

a3s_boot/app/
microservice.rs
