sabi/tokio/data_src/
global_setup.rs

1// Copyright (C) 2024-2026 Takayuki Sato. All Rights Reserved.
2// This program is free software under MIT License.
3// See the file LICENSE in this distribution for more details.
4
5use super::DataSrcError;
6
7use super::super::{
8    AutoShutdown, DataConn, DataConnContainer, DataSrc, DataSrcContainer, DataSrcManager,
9    StaticDataSrcContainer, StaticDataSrcRegistration,
10};
11use crate::SendSyncNonNull;
12
13use setup_read_cleanup::{PhasedCellAsync, PhasedError, PhasedErrorKind};
14
15use std::collections::HashMap;
16use std::sync::Arc;
17use std::{any, ptr};
18use tokio::sync::Mutex;
19
20pub(crate) static DS_MANAGER: PhasedCellAsync<DataSrcManager> =
21    PhasedCellAsync::new(DataSrcManager::new(false));
22
23const NOOP: fn(&mut DataSrcManager) -> Result<(), PhasedError> = |_| Ok(());
24
25impl Drop for AutoShutdown {
26    fn drop(&mut self) {
27        let _ = DS_MANAGER.force_to_cleanup(|ds_m| {
28            ds_m.close();
29            Ok::<(), PhasedError>(())
30        });
31    }
32}
33
34/// Registers a global data source, making it available throughout the application.
35///
36/// Global data sources are managed by a singleton and can be set up once for the application's
37/// lifetime.
38/// If `setup_async` or `setup_with_order_async` has already been called, this function will return
39/// an `errs::Err`.
40/// If another Tokio task holds the lock of the global data source manager, this function will wait
41/// until the lock is released.
42///
43/// # Parameters
44///
45/// * `name` - The name to associate with this data source.
46/// * `ds` - The data source instance, which must implement `DataSrc` and have a `'static` lifetime.
47///
48/// # Type Parameters
49///
50/// * `S` - The type of the data source.
51/// * `C` - The type of the data connection provided by the data source.
52///
53/// # Returns
54///
55/// * `errs::Result<()>`: [`Ok`] if the data source is successfully registered, or an [`errs::Err`] if
56///   the global data source manager is in an invalid state or setup has already occurred.
57pub async fn uses_async<S, C>(name: impl Into<Arc<str>>, ds: S) -> errs::Result<()>
58where
59    S: DataSrc<C> + 'static,
60    C: DataConn + 'static,
61{
62    match DS_MANAGER.lock_async().await {
63        Ok(mut dsm) => {
64            dsm.add(name, ds);
65            Ok(())
66        }
67        Err(e) => Err(errs::Err::with_source(
68            DataSrcError::FailToRegisterGlobalDataSrc { name: name.into() },
69            e,
70        )),
71    }
72}
73
74/// Registers a global data source, making it available throughout the application.
75///
76/// This is the synchronous version of `uses_async`.
77/// Global data sources are managed by a singleton and can be set up once for the application's
78/// lifetime.
79/// If `setup_async` or `setup_with_order_async` has already been called, this function will return
80/// an `errs::Err`.
81/// If another Tokio task holds the lock of the global data source manager, this function will return
82/// an error immediately without waiting.
83///
84/// # Parameters
85///
86/// * `name` - The name to associate with this data source.
87/// * `ds` - The data source instance, which must implement `DataSrc` and have a `'static` lifetime.
88///
89/// # Type Parameters
90///
91/// * `S` - The type of the data source.
92/// * `C` - The type of the data connection provided by the data source.
93///
94/// # Returns
95///
96/// * `errs::Result<()>`: [`Ok`] if the data source is successfully registered, or an [`errs::Err`] if
97///   the global data source manager is in an invalid state or setup has already occurred.
98pub fn uses<S, C>(name: impl Into<Arc<str>>, ds: S) -> errs::Result<()>
99where
100    S: DataSrc<C> + 'static,
101    C: DataConn + 'static,
102{
103    match DS_MANAGER.try_lock() {
104        Ok(mut dsm) => {
105            dsm.add(name, ds);
106            Ok(())
107        }
108        Err(e) => Err(errs::Err::with_source(
109            DataSrcError::FailToRegisterGlobalDataSrc { name: name.into() },
110            e,
111        )),
112    }
113}
114
115fn collect_static_data_src_containers(dsm: &mut DataSrcManager) {
116    let regs: Vec<_> = inventory::iter::<StaticDataSrcRegistration>
117        .into_iter()
118        .collect();
119
120    let mut static_vec: Vec<SendSyncNonNull<DataSrcContainer>> = Vec::with_capacity(regs.len());
121    for reg in regs {
122        let any_container = (reg.factory)();
123        static_vec.push(any_container.ssnnptr);
124    }
125
126    dsm.prepend(static_vec);
127}
128
129/// Asynchronously sets up all globally registered data sources.
130///
131/// This function sets up all data sources that have been registered via `uses_async` function or
132/// `uses!` macro. It collects any setup errors.
133///
134/// # Returns
135///
136/// A `Result` which is `Ok` containing an `AutoShutdown` guard if all data sources
137/// are set up successfully. If setup fails for any data source, it returns an `Err`
138/// with `DataSrcError::FailToSetupGlobalDataSrcs`. If called when data sources are
139/// already set up or in transition, it returns `DataSrcError::AlreadySetupGlobalDataSrcs`
140/// or `DataSrcError::DuringSetupGlobalDataSrcs` respectively.
141pub async fn setup_async() -> errs::Result<AutoShutdown> {
142    let errors = Arc::new(Mutex::new(Vec::new()));
143    let errors_for_closure = Arc::clone(&errors);
144
145    if let Err(e) = DS_MANAGER
146        .transition_to_read_async(move |ds_m| {
147            collect_static_data_src_containers(ds_m);
148            let errors_for_future = Arc::clone(&errors_for_closure);
149            Box::pin(async move {
150                let mut lock = errors_for_future.lock().await;
151                ds_m.setup_async(&mut lock).await;
152                Ok::<(), PhasedError>(())
153            })
154        })
155        .await
156    {
157        if e.kind() == PhasedErrorKind::DuringTransitionToRead {
158            return Err(errs::Err::new(DataSrcError::DuringSetupGlobalDataSrcs));
159        } else {
160            return Err(errs::Err::new(DataSrcError::AlreadySetupGlobalDataSrcs));
161        }
162    }
163
164    let errors = Arc::try_unwrap(errors).unwrap().into_inner();
165    if errors.is_empty() {
166        Ok(AutoShutdown {})
167    } else {
168        Err(errs::Err::new(DataSrcError::FailToSetupGlobalDataSrcs {
169            errors,
170        }))
171    }
172}
173
174/// Asynchronously sets up all globally registered data sources with a specified order.
175///
176/// Similar to `setup_async`, but allows defining the order in which data sources
177/// are set up. Data sources not specified in `names` will be set up after the
178/// specified ones, in an undefined order.
179///
180/// # Parameters
181///
182/// * `names` - An array of string slices specifying the desired setup order by data source name.
183///
184/// # Returns
185///
186/// A `Result` which is `Ok` containing an `AutoShutdown` guard if all data sources
187/// are set up successfully. If setup fails for any data source, it returns an `Err`
188/// with `DataSrcError::FailToSetupGlobalDataSrcs`. If called when data sources are
189/// already set up or in transition, it returns `DataSrcError::AlreadySetupGlobalDataSrcs`
190/// or `DataSrcError::DuringSetupGlobalDataSrcs` respectively.
191pub async fn setup_with_order_async(names: &'static [&str]) -> errs::Result<AutoShutdown> {
192    let errors = Arc::new(Mutex::new(Vec::new()));
193    let errors_for_closure = Arc::clone(&errors);
194
195    if let Err(e) = DS_MANAGER
196        .transition_to_read_async(move |ds_m| {
197            collect_static_data_src_containers(ds_m);
198            let errors_for_future = Arc::clone(&errors_for_closure);
199            Box::pin(async move {
200                let mut lock = errors_for_future.lock().await;
201                ds_m.setup_with_order_async(names, &mut lock).await;
202                Ok::<(), PhasedError>(())
203            })
204        })
205        .await
206    {
207        if e.kind() == PhasedErrorKind::DuringTransitionToRead {
208            return Err(errs::Err::new(DataSrcError::DuringSetupGlobalDataSrcs));
209        } else {
210            return Err(errs::Err::new(DataSrcError::AlreadySetupGlobalDataSrcs));
211        }
212    }
213
214    let errors = Arc::try_unwrap(errors).unwrap().into_inner();
215    if errors.is_empty() {
216        Ok(AutoShutdown {})
217    } else {
218        Err(errs::Err::new(DataSrcError::FailToSetupGlobalDataSrcs {
219            errors,
220        }))
221    }
222}
223
224pub(crate) fn copy_global_data_srcs_to_map(index_map: &mut HashMap<Arc<str>, (bool, usize)>) {
225    if let Ok(ds_m) = DS_MANAGER.read_relaxed() {
226        ds_m.copy_ds_ready_to_map(index_map);
227    } else if (match DS_MANAGER.force_to_read(NOOP) {
228        Ok(_) => Ok(()),
229        Err(e) => match e.kind() {
230            PhasedErrorKind::PhaseIsAlreadyCleanup => Ok(()),
231            PhasedErrorKind::DuringTransitionToRead => Ok(()),
232            _ => Err(()),
233        },
234    })
235    .is_ok()
236    {
237        if let Ok(ds_m) = DS_MANAGER.read_relaxed() {
238            ds_m.copy_ds_ready_to_map(index_map);
239        }
240    }
241}
242
243#[doc(hidden)]
244pub fn create_static_data_src_container<S, C>(
245    name: &'static str,
246    data_src: S,
247) -> StaticDataSrcContainer
248where
249    S: DataSrc<C> + 'static,
250    C: DataConn + 'static,
251{
252    let boxed = Box::new(DataSrcContainer::<S, C>::new(name, data_src, false));
253    let ptr = ptr::NonNull::from(Box::leak(boxed)).cast::<DataSrcContainer>();
254    StaticDataSrcContainer {
255        ssnnptr: SendSyncNonNull::new(ptr),
256    }
257}
258
259impl StaticDataSrcRegistration {
260    pub const fn new(factory: fn() -> StaticDataSrcContainer) -> Self {
261        Self { factory }
262    }
263}
264inventory::collect!(StaticDataSrcRegistration);
265
266#[macro_export]
267#[doc(hidden)]
268macro_rules! _uses_for_async {
269    ($name:tt, $data_src:expr) => {
270        const _: () = {
271            inventory::submit! {
272                $crate::tokio::StaticDataSrcRegistration::new(|| {
273                    $crate::tokio::create_static_data_src_container($name, $data_src)
274                })
275            }
276        };
277    };
278}
279
280pub(crate) async fn create_data_conn_from_global_data_src_async<C>(
281    index: usize,
282    name: impl AsRef<str>,
283) -> errs::Result<Box<DataConnContainer>>
284where
285    C: DataConn + 'static,
286{
287    match DS_MANAGER.read_relaxed() {
288        Ok(ds_manager) => ds_manager.create_data_conn_async::<C>(index, name).await,
289        Err(e) => Err(errs::Err::with_source(
290            DataSrcError::FailToCreateDataConn {
291                name: name.as_ref().into(),
292                data_conn_type: any::type_name::<C>(),
293            },
294            e,
295        )),
296    }
297}
sabi/tokio/data_src/global_setup.rs

sabi/tokio/data_src/
global_setup.rs
