fusion_core/
application.rs

1use std::{
2  any::Any,
3  fmt::Display,
4  future::Future,
5  sync::{Arc, OnceLock},
6};
7
8use config::Config;
9use dashmap::DashMap;
10use fusion_common::ahash::HashSet;
11use log::{debug, info};
12use mea::{
13  mutex::Mutex,
14  shutdown::{ShutdownRecv, ShutdownSend},
15};
16use serde::de::DeserializeOwned;
17
18use fusion_common::time::OffsetDateTime;
19
20use crate::{
21  Result,
22  component::{ComponentArc, ComponentError, ComponentResult, DynComponentArc, auto_inject_component},
23  configuration::{ConfigRegistry, Configurable, ConfigureResult, FusionConfigRegistry, FusionSetting},
24  log::LogPlugin,
25  plugin::{Plugin, PluginRef},
26};
27
28type Registry<T> = DashMap<String, T>;
29type Task<T> = dyn FnOnce(Application) -> Box<dyn Future<Output = Result<T>> + Send>;
30
31pub(crate) struct ApplicationInner {
32  config_registry: FusionConfigRegistry,
33  components: Registry<DynComponentArc>,
34  start_time: OffsetDateTime,
35  pub(crate) shutdown: Mutex<Option<(ShutdownSend, ShutdownRecv)>>,
36}
37
38/// Application, clone is cheap.
39#[derive(Clone)]
40pub struct Application(pub(crate) Arc<ApplicationInner>);
41
42impl Display for Application {
43  fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
44    write!(f, "Application({}|{})", self.fusion_setting().app().name(), self.0.start_time)
45  }
46}
47
48impl Application {
49  pub fn builder() -> ApplicationBuilder {
50    ApplicationBuilder::default()
51  }
52
53  pub fn global() -> Application {
54    GLOBAL_APPLICATION.get().expect("Application is not initialized").clone()
55  }
56
57  pub fn set_global(application: Application) {
58    match GLOBAL_APPLICATION.set(application) {
59      Ok(_) => (),
60      Err(old) => {
61        panic!("Global application was already set to {}", old)
62      }
63    }
64  }
65
66  pub async fn shutdown() {
67    let app = Self::global();
68    if let Some((shutdown_tx, _)) = app.0.shutdown.lock().await.as_ref() {
69      shutdown_tx.shutdown();
70    }
71  }
72
73  pub async fn await_shutdown() -> bool {
74    let app = Self::global();
75    if let Some((shutdown_tx, shutdown_rx)) = app.0.shutdown.lock().await.take() {
76      drop(shutdown_rx);
77      shutdown_tx.await_shutdown().await;
78      true
79    } else {
80      false
81    }
82  }
83
84  pub async fn is_shutdown(&self) -> bool {
85    self.0.shutdown.lock().await.is_none()
86  }
87
88  pub async fn shutdown_recv(&self) -> ShutdownRecv {
89    let maybe = self.0.shutdown.lock().await;
90    let tuple = maybe.as_ref().unwrap();
91    tuple.1.clone()
92  }
93
94  pub fn config_registry(&self) -> &FusionConfigRegistry {
95    &self.0.config_registry
96  }
97
98  /// Get the component reference of the specified type
99  #[inline]
100  pub fn get_component_arc<T>(&self) -> ComponentResult<ComponentArc<T>>
101  where
102    T: Any + Send + Sync,
103  {
104    self.get_component_ref_by_name(std::any::type_name::<T>())
105  }
106
107  pub fn component_arc<T>(&self) -> ComponentArc<T>
108  where
109    T: Any + Send + Sync,
110  {
111    let component_name = std::any::type_name::<T>();
112    match self.get_component_ref_by_name(component_name) {
113      Ok(c) => c,
114      Err(e) => panic!("{:?}", e),
115    }
116  }
117
118  pub fn get_component_ref_by_name<T>(&self, component_name: &str) -> ComponentResult<ComponentArc<T>>
119  where
120    T: Any + Send + Sync,
121  {
122    let pair = match self.0.components.get(component_name) {
123      Some(pair) => pair,
124      None => return Err(ComponentError::ComponentNotFound(component_name.to_string())),
125    };
126    let component_ref = pair.value().clone();
127    component_ref.downcast::<T>()
128  }
129
130  pub fn component<T>(&self) -> T
131  where
132    T: Clone + Send + Sync + 'static,
133  {
134    match self.get_component() {
135      Ok(c) => c,
136      Err(e) => panic!("{:?}", e),
137    }
138  }
139
140  /// Get the component of the specified type
141  pub fn get_component<T>(&self) -> ComponentResult<T>
142  where
143    T: Clone + Send + Sync + 'static,
144  {
145    let component_ref = self.get_component_arc();
146    component_ref.map(|c| T::clone(&c))
147  }
148
149  /// Get all built components. The return value is the full crate path of all components
150  pub fn get_component_names(&self) -> Vec<String> {
151    self.0.components.iter().map(|e| e.key().clone()).collect()
152  }
153
154  pub fn add_component<T>(&self, component: T)
155  where
156    T: Clone + Any + Send + Sync,
157  {
158    let component_name = std::any::type_name::<T>();
159    if self.0.components.contains_key(component_name) {
160      panic!("Error adding component {component_name}: component was already added in application")
161    }
162    self.0.components.insert(component_name.to_string(), DynComponentArc::new(component));
163
164    debug!("added component: {}", component_name);
165  }
166
167  pub fn fusion_setting(&self) -> Arc<FusionSetting> {
168    self.0.config_registry.fusion_setting()
169  }
170
171  /// Get `::config::Config` Instance
172  pub fn underlying_config(&self) -> Arc<Config> {
173    self.0.config_registry.config()
174  }
175
176  pub fn start_time(&self) -> &OffsetDateTime {
177    &self.0.start_time
178  }
179}
180
181impl ConfigRegistry for Application {
182  fn get_config<T>(&self) -> ConfigureResult<T>
183  where
184    T: DeserializeOwned + Configurable,
185  {
186    self.0.config_registry.get_config()
187  }
188
189  fn get_config_by_path<T>(&self, path: &str) -> ConfigureResult<T>
190  where
191    T: DeserializeOwned,
192  {
193    self.0.config_registry.get_config_by_path(path)
194  }
195}
196
197static GLOBAL_APPLICATION: OnceLock<Application> = OnceLock::new();
198
199#[derive(Default)]
200pub struct ApplicationBuilder {
201  config_registry: FusionConfigRegistry,
202
203  /// Plugins
204  pub(crate) plugin_registry: Registry<PluginRef>,
205
206  /// Components
207  pub(crate) components: Registry<DynComponentArc>,
208
209  /// Tasks
210  shutdown_hooks: Vec<Box<Task<String>>>,
211}
212
213unsafe impl Send for ApplicationBuilder {}
214unsafe impl Sync for ApplicationBuilder {}
215
216impl ApplicationBuilder {
217  pub fn get_fusion_config(&self) -> Arc<FusionSetting> {
218    self.config_registry.fusion_setting()
219  }
220
221  pub fn with_config_registry(&mut self, config_registry: FusionConfigRegistry) -> &Self {
222    self.config_registry = config_registry;
223    self
224  }
225
226  /// add Config Source
227  pub fn add_config_source<T>(&mut self, source: T) -> &mut Self
228  where
229    T: config::Source + Send + Sync + 'static,
230  {
231    self.config_registry.add_config_source(source).expect("Add config source failed");
232    self
233  }
234
235  /// add plugin
236  pub fn add_plugin<T: Plugin>(&mut self, plugin: T) -> &mut Self {
237    let plugin_name = plugin.name().to_string();
238    debug!("added plugin: {plugin_name}");
239
240    if plugin.immediately() {
241      plugin.immediately_build(self);
242      return self;
243    }
244    if self.plugin_registry.contains_key(plugin.name()) {
245      panic!("Error adding plugin {plugin_name}: plugin was already added in application")
246    }
247    self.plugin_registry.insert(plugin_name, PluginRef::new(plugin));
248    self
249  }
250
251  /// Returns `true` if the [`Plugin`] has already been added.
252  #[inline]
253  pub fn contains_plugin<T: Plugin>(&self) -> bool {
254    self.plugin_registry.contains_key(std::any::type_name::<T>())
255  }
256
257  /// Add component to the registry
258  pub fn add_component<T>(&mut self, component: T) -> &mut Self
259  where
260    T: Clone + Any + Send + Sync,
261  {
262    let component_name = std::any::type_name::<T>();
263    if self.components.contains_key(component_name) {
264      panic!("Error adding component {component_name}: component was already added in application builder")
265    }
266    self.components.insert(component_name.to_string(), DynComponentArc::new(component));
267
268    debug!("added component: {}", component_name);
269    self
270  }
271
272  /// Get the component of the specified type
273  pub fn get_component_ref<T>(&self) -> ComponentResult<ComponentArc<T>>
274  where
275    T: Any + Send + Sync,
276  {
277    let component_name = std::any::type_name::<T>();
278    let pair = match self.components.get(component_name) {
279      Some(pair) => pair,
280      None => return Err(ComponentError::ComponentNotFound(component_name.to_string())),
281    };
282    let component_ref = pair.value().clone();
283    component_ref.downcast::<T>()
284  }
285
286  pub fn component<T>(&self) -> T
287  where
288    T: Clone + Send + Sync + 'static,
289  {
290    match self.get_component() {
291      Ok(c) => c,
292      Err(e) => panic!("{:?}", e),
293    }
294  }
295
296  /// get cloned component
297  pub fn get_component<T>(&self) -> ComponentResult<T>
298  where
299    T: Clone + Send + Sync + 'static,
300  {
301    let component_ref = self.get_component_ref();
302    component_ref.map(|c| T::clone(&c))
303  }
304
305  /// Add a shutdown hook
306  pub fn add_shutdown_hook<T>(&mut self, hook: T) -> &mut Self
307  where
308    T: FnOnce(Application) -> Box<dyn Future<Output = Result<String>> + Send> + 'static,
309  {
310    self.shutdown_hooks.push(Box::new(hook));
311    self
312  }
313
314  /// The `run` method is suitable for applications that contain scheduling logic,
315  /// such as web, job, and stream.
316  ///
317  pub async fn run(&mut self) -> Result<Application> {
318    self.inner_run().await
319  }
320
321  async fn inner_run(&mut self) -> Result<Application> {
322    let app = self.build().await?;
323
324    // 4. schedule
325    // self.schedule().await
326
327    Ok(app)
328  }
329
330  /// Unlike the [`run`] method, the `build` method is suitable for applications that do not contain scheduling logic.
331  /// This method returns the built Application, and developers can implement logic such as command lines and task scheduling by themselves.
332  pub async fn build(&mut self) -> Result<Application> {
333    // 0. load toml config
334    // self.load_config_if_need()?;
335
336    // 1. build plugin
337    self.build_plugins().await?;
338
339    // 2. service dependency inject
340    auto_inject_component(self)?;
341
342    // 3. build application
343    let application = self.build_application();
344
345    Application::set_global(application);
346    Ok(Application::global())
347  }
348
349  /// Initialize tracing for Application
350  async fn build_plugins(&mut self) -> Result<()> {
351    self.add_plugin(LogPlugin);
352
353    let registry = std::mem::take(&mut self.plugin_registry);
354    let mut to_register = registry.iter().map(|e| e.value().to_owned()).collect::<Vec<_>>();
355    let mut registered: HashSet<String> = HashSet::default();
356
357    while !to_register.is_empty() {
358      let mut progress = false;
359      let mut next_round = vec![];
360
361      for plugin in to_register {
362        let deps = plugin.dependencies();
363        if deps.iter().all(|dep| registered.contains(*dep)) {
364          plugin.build(self).await;
365          registered.insert(plugin.name().to_string());
366          info!("{} plugin registered", plugin.name());
367          progress = true;
368        } else {
369          next_round.push(plugin);
370        }
371      }
372
373      if !progress {
374        panic!("Cyclic dependency detected or missing dependencies for some plugins");
375      }
376
377      to_register = next_round;
378    }
379
380    self.plugin_registry = registry;
381    Ok(())
382  }
383
384  fn build_application(&mut self) -> Application {
385    let components = std::mem::take(&mut self.components);
386    let configuration_state = std::mem::take(&mut self.config_registry);
387    let init_time = configuration_state.fusion_setting().app().time_now();
388    let shutdown = Mutex::new(Some(mea::shutdown::new_pair()));
389    Application(Arc::new(ApplicationInner {
390      config_registry: configuration_state,
391      components,
392      start_time: init_time,
393      shutdown,
394    }))
395  }
396}
397
398impl ConfigRegistry for ApplicationBuilder {
399  fn get_config<T>(&self) -> ConfigureResult<T>
400  where
401    T: DeserializeOwned + Configurable,
402  {
403    self.config_registry.get_config::<T>()
404  }
405
406  fn get_config_by_path<T>(&self, path: &str) -> ConfigureResult<T>
407  where
408    T: DeserializeOwned,
409  {
410    self.config_registry.get_config_by_path::<T>(path)
411  }
412}
413#[cfg(test)]
414mod tests {
415  use fusion_common::env::remove_envs;
416
417  use super::*;
418
419  #[tokio::test]
420  async fn test_application_run() {
421    remove_envs(&[
422      "FUSION__APP__NAME",
423      "FUSION__WEB__SERVER_ADDR",
424      "FUSION__SECURITY__TOKEN__SECRET_KEY",
425      "FUSION__SECURITY__PWD__PWD_KEY",
426    ])
427    .unwrap();
428    Application::builder().run().await.unwrap();
429    let app = Application::global();
430    assert_eq!(app.fusion_setting().app().name(), "fusion");
431  }
432}
fusion_core/application.rs

fusion_core/
application.rs
