sabi/
data_hub.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 crate::data_src::{copy_global_data_srcs_to_map, create_data_conn_from_global_data_src};
6use crate::{DataConn, DataConnManager, DataHub, DataSrc, DataSrcManager, SendSyncNonNull};
7
8#[allow(unused)] // for rustdoc
9use crate::DataAcc;
10
11use crate::{DataConnContainer, ErrEntry};
12
13use std::collections::HashMap;
14use std::sync::Arc;
15use std::{any, ptr};
16
17/// An enum type representing the reasons for errors that can occur within [`DataHub`] operations.
18#[derive(Debug)]
19pub enum DataHubError {
20    /// Indicates a failure during the setup process of one or more session-local data sources.
21    /// Contains a vector of data source names and their corresponding errors.
22    FailToSetupLocalDataSrcs {
23        /// The vector contains errors that occurred in each [`DataSrc`] object.
24        errors: Vec<ErrEntry>,
25    },
26
27    /// Indicates that no [`DataSrc`] was found to create a [`DataConn`] for the specified name
28    /// and type.
29    NoDataSrcToCreateDataConn {
30        /// The name of the data source that could not be found.
31        name: Arc<str>,
32
33        /// The type name of the [`DataConn`] that was requested.
34        data_conn_type: &'static str,
35    },
36}
37
38impl DataHub {
39    /// Creates a new [`DataHub`] instance.
40    ///
41    /// Upon creation, it collects references to globally set-up data sources
42    /// into its internal map for quick access.
43    #[allow(clippy::new_without_default)]
44    pub fn new() -> Self {
45        let mut data_src_map = HashMap::new();
46        copy_global_data_srcs_to_map(&mut data_src_map);
47
48        Self {
49            local_data_src_manager: DataSrcManager::new(true),
50            data_src_map,
51            data_conn_manager: DataConnManager::new(),
52            fixed: false,
53        }
54    }
55
56    /// Creates a new [`DataHub`] instance with a specified commit order for data connections.
57    ///
58    /// This constructor allows defining a specific order for pre-commit, commit, and post-commit
59    /// operations for named data connections. Data connections not specified in `names` will
60    /// be processed after the named ones, in their order of acquisition.
61    ///
62    /// Upon creation, it collects references to globally set-up data sources
63    /// into its internal map for quick access.
64    ///
65    /// # Parameters
66    ///
67    /// * `names`: A slice of `&str` representing the names of data connections to commit in a
68    ///   specific order.
69    pub fn with_commit_order(names: &[&str]) -> Self {
70        let mut data_src_map = HashMap::new();
71        copy_global_data_srcs_to_map(&mut data_src_map);
72
73        Self {
74            local_data_src_manager: DataSrcManager::new(true),
75            data_src_map,
76            data_conn_manager: DataConnManager::with_commit_order(names),
77            fixed: false,
78        }
79    }
80
81    /// Registers a session-local data source with this [`DataHub`] instance.
82    ///
83    /// This method is similar to the global [`uses!`] macro but registers a data source
84    /// that is local to this specific [`DataHub`] session. Once the [`DataHub`]'s state is
85    /// "fixed" (while [`DataHub::run`] or [`DataHub::txn`] method is executing),
86    /// further calls to `uses` are ignored. However, after the method completes,
87    /// the [`DataHub`]'s "fixed" state is reset, allowing for new data sources to be
88    /// registered or removed via [`DataHub::disuses`] method in subsequent operations.
89    ///
90    /// # Parameters
91    ///
92    /// * `name`: The unique name for the local data source.
93    /// * `ds`: The [`DataSrc`] instance to register.
94    #[allow(rustdoc::broken_intra_doc_links)]
95    pub fn uses<S, C>(&mut self, name: impl Into<Arc<str>>, ds: S)
96    where
97        S: DataSrc<C>,
98        C: DataConn + 'static,
99    {
100        if self.fixed {
101            return;
102        }
103        self.local_data_src_manager.add(name, ds);
104    }
105
106    /// Unregisters and drops a session-local data source by its name.
107    ///
108    /// This method removes a data source that was previously registered via [`DataHub::uses`].
109    /// This operation is ignored if the [`DataHub`]'s state is already "fixed".
110    ///
111    /// # Parameters
112    ///
113    /// * `name`: The name of the local data source to unregister.
114    pub fn disuses(&mut self, name: impl AsRef<str>) {
115        if self.fixed {
116            return;
117        }
118        self.data_src_map.remove(name.as_ref());
119        self.local_data_src_manager.remove(name);
120    }
121
122    #[inline]
123    fn begin(&mut self) -> errs::Result<()> {
124        self.fixed = true;
125
126        let mut errors = Vec::new();
127
128        self.local_data_src_manager.setup(&mut errors);
129        if errors.is_empty() {
130            self.local_data_src_manager
131                .copy_ds_ready_to_map(&mut self.data_src_map);
132            Ok(())
133        } else {
134            Err(errs::Err::new(DataHubError::FailToSetupLocalDataSrcs {
135                errors,
136            }))
137        }
138    }
139
140    #[inline]
141    fn end(&mut self) {
142        self.data_conn_manager.close();
143        self.fixed = false;
144    }
145
146    /// Executes a given logic function without transaction control.
147    ///
148    /// This method sets up local data sources, runs the provided closure,
149    /// and then cleans up the [`DataHub`]'s session resources. It does not
150    /// perform commit or rollback operations.
151    ///
152    /// # Parameters
153    ///
154    /// * `logic_fn`: A closure that encapsulates the business logic to be executed.
155    ///   It takes a mutable reference to [`DataHub`] as an argument.
156    ///
157    /// # Returns
158    ///
159    /// * `errs::Result<()>`: The result of the logic function's execution,
160    ///   or an error if executing `logic_fn` fails.
161    pub fn run<F>(&mut self, mut logic_fn: F) -> errs::Result<()>
162    where
163        F: FnMut(&mut DataHub) -> errs::Result<()>,
164    {
165        let mut r = self.begin();
166        if r.is_ok() {
167            r = logic_fn(self);
168        }
169        self.end();
170        r
171    }
172
173    /// Executes a given logic function within a managed transaction.
174    ///
175    /// This method starts by setting up local data sources, runs the provided closure,
176    /// and then attempts to commit all open data connections in the session.
177    ///
178    /// If any error occurs during the execution of the closure or during the commit phase,
179    /// it initiates a rollback on all data connections and reports the transaction failure details.
180    /// Finally, it cleans up session resources.
181    ///
182    /// # Parameters
183    ///
184    /// * `logic_fn`: A closure that encapsulates the business logic to be executed.
185    ///   It takes a mutable reference to [`DataHub`] as an argument.
186    ///
187    /// # Returns
188    ///
189    /// * `errs::Result<()>`: `Ok(())` if the closure and the commit phase succeed,
190    ///   or an [`errs::Err`] if any phase fails.
191    pub fn txn<F>(&mut self, mut logic_fn: F) -> errs::Result<()>
192    where
193        F: FnMut(&mut DataHub) -> errs::Result<()>,
194    {
195        let mut r = self.begin();
196        if r.is_ok() {
197            r = logic_fn(self);
198        }
199
200        let mut reports = self.data_conn_manager.new_failure_reports();
201
202        if r.is_ok() {
203            r = self.data_conn_manager.commit(&mut reports);
204        }
205        if r.is_err() {
206            self.data_conn_manager.rollback(reports);
207        }
208
209        self.end();
210        r
211    }
212
213    /// Retrieves a mutable reference to a [`DataConn`] object by name, creating it if necessary.
214    ///
215    /// This is the core method used by [`DataAcc`] implementations to obtain connections
216    /// to external data services. It first checks if a [`DataConn`] with the given name
217    /// already exists in the [`DataHub`]'s session. If not, it attempts to find a
218    /// corresponding [`DataSrc`] and create a new [`DataConn`] from it.
219    ///
220    /// # Type Parameters
221    ///
222    /// * `C`: The concrete type of [`DataConn`] expected.
223    ///
224    /// # Parameters
225    ///
226    /// * `name`: The name of the data source/connection to retrieve.
227    ///
228    /// # Returns
229    ///
230    /// * `errs::Result<&mut C>`: A mutable reference to the [`DataConn`] instance if successful,
231    ///   or an [`errs::Err`] if the data source is not found, or if the retrieved/created
232    ///   [`DataConn`] cannot be cast to the specified type `C`.
233    pub fn get_data_conn<C>(&mut self, name: &str) -> errs::Result<&mut C>
234    where
235        C: DataConn + 'static,
236    {
237        if let Some(ssnnptr) = self.data_conn_manager.find_by_name(name) {
238            let typed_ssnnptr = DataConnManager::to_typed_ptr::<C>(&ssnnptr)?;
239            return Ok(unsafe { &mut (*typed_ssnnptr).data_conn });
240        }
241
242        if let Some((local, index)) = self.data_src_map.get(name) {
243            let boxed = if *local {
244                self.local_data_src_manager
245                    .create_data_conn::<C>(*index, name)?
246            } else {
247                create_data_conn_from_global_data_src::<C>(*index, name)?
248            };
249
250            let ptr = Box::into_raw(boxed);
251            if let Some(nnptr) = ptr::NonNull::new(ptr) {
252                let ssnnptr = SendSyncNonNull::new(nnptr);
253                self.data_conn_manager.add(ssnnptr);
254
255                let typed_ptr = ptr.cast::<DataConnContainer<C>>();
256                return Ok(unsafe { &mut (*typed_ptr).data_conn });
257            } else {
258                // impossible case.
259            }
260        }
261
262        Err(errs::Err::new(DataHubError::NoDataSrcToCreateDataConn {
263            name: name.into(),
264            data_conn_type: any::type_name::<C>(),
265        }))
266    }
267}
268
269#[cfg_attr(coverage_nightly, coverage(off))]
270#[cfg(test)]
271mod tests_of_data_hub {
272    use super::*;
273    use crate::{AsyncGroup, DataConnError, DataSrcError, TxnFailureReport};
274    use std::sync::Mutex;
275
276    #[derive(Clone, Copy, PartialEq)]
277    enum Failure {
278        None,
279        FailToPreCommit,
280        FailToCommit,
281        FailToPostCommit,
282        FailToRollback,
283        FailToSetup,
284        FailToCreateDataConn,
285    }
286
287    struct MyDataConn {
288        id: i8,
289        failure: Failure,
290        committed: bool,
291        logger: Arc<Mutex<Vec<String>>>,
292    }
293    impl MyDataConn {
294        fn new(id: i8, logger: Arc<Mutex<Vec<String>>>, failure: Failure) -> Self {
295            logger
296                .lock()
297                .unwrap()
298                .push(format!("MyDataConn::new {}", id));
299            Self {
300                id,
301                failure,
302                committed: false,
303                logger,
304            }
305        }
306    }
307    impl Drop for MyDataConn {
308        fn drop(&mut self) {
309            self.logger
310                .lock()
311                .unwrap()
312                .push(format!("MyDataConn::drop {}", self.id));
313        }
314    }
315    impl DataConn for MyDataConn {
316        fn pre_commit(&mut self, _ag: &mut AsyncGroup) -> errs::Result<()> {
317            if self.failure == Failure::FailToPreCommit {
318                self.logger
319                    .lock()
320                    .unwrap()
321                    .push(format!("MyDataConn::pre_commit {} failed", self.id));
322                Err(errs::Err::new("pre commit error"))
323            } else {
324                self.logger
325                    .lock()
326                    .unwrap()
327                    .push(format!("MyDataConn::pre_commit {}", self.id));
328                Ok(())
329            }
330        }
331        fn commit(&mut self, _ag: &mut AsyncGroup) -> errs::Result<()> {
332            if self.failure == Failure::FailToCommit {
333                self.logger
334                    .lock()
335                    .unwrap()
336                    .push(format!("MyDataConn::commit {} failed", self.id));
337                Err(errs::Err::new("commit error"))
338            } else {
339                self.logger
340                    .lock()
341                    .unwrap()
342                    .push(format!("MyDataConn::commit {}", self.id));
343                self.committed = true;
344                Ok(())
345            }
346        }
347        fn is_committed(&self) -> bool {
348            false
349        }
350        fn post_commit(&mut self, _ag: &mut AsyncGroup) -> errs::Result<()> {
351            if self.failure == Failure::FailToPostCommit {
352                self.logger
353                    .lock()
354                    .unwrap()
355                    .push(format!("MyDataConn::post_commit {} failed", self.id));
356                Err(errs::Err::new("post commit error"))
357            } else {
358                self.logger
359                    .lock()
360                    .unwrap()
361                    .push(format!("MyDataConn::post_commit {}", self.id));
362                Ok(())
363            }
364        }
365        fn rollback(&mut self, _ag: &mut AsyncGroup) -> errs::Result<()> {
366            if self.failure == Failure::FailToRollback {
367                self.logger
368                    .lock()
369                    .unwrap()
370                    .push(format!("MyDataConn::rollback {} failed", self.id));
371                Err(errs::Err::new("rollback error"))
372            } else {
373                self.logger
374                    .lock()
375                    .unwrap()
376                    .push(format!("MyDataConn::rollback {}", self.id));
377                Ok(())
378            }
379        }
380        fn on_txn_failure(&mut self, _ag: &mut AsyncGroup, _reports: &[TxnFailureReport]) {
381            self.logger
382                .lock()
383                .unwrap()
384                .push(format!("MyDataConn::on_txn_failure {}", self.id));
385        }
386        fn close(&mut self) {
387            self.logger
388                .lock()
389                .unwrap()
390                .push(format!("MyDataConn::close {}", self.id));
391        }
392    }
393
394    struct MyDataSrc {
395        id: i8,
396        failure: Failure,
397        logger: Arc<Mutex<Vec<String>>>,
398    }
399    impl MyDataSrc {
400        fn new(id: i8, logger: Arc<Mutex<Vec<String>>>, failure: Failure) -> Self {
401            logger
402                .lock()
403                .unwrap()
404                .push(format!("MyDataSrc::new {}", id));
405            Self {
406                id,
407                failure,
408                logger,
409            }
410        }
411    }
412    impl Drop for MyDataSrc {
413        fn drop(&mut self) {
414            self.logger
415                .lock()
416                .unwrap()
417                .push(format!("MyDataSrc::drop {}", self.id));
418        }
419    }
420    impl DataSrc<MyDataConn> for MyDataSrc {
421        fn setup(&mut self, _ag: &mut AsyncGroup) -> errs::Result<()> {
422            if self.failure == Failure::FailToSetup {
423                self.logger
424                    .lock()
425                    .unwrap()
426                    .push(format!("MyDataSrc::setup {} failed", self.id));
427                Err(errs::Err::new("setup error".to_string()))
428            } else {
429                self.logger
430                    .lock()
431                    .unwrap()
432                    .push(format!("MyDataSrc::setup {}", self.id));
433                Ok(())
434            }
435        }
436        fn close(&mut self) {
437            self.logger
438                .lock()
439                .unwrap()
440                .push(format!("MyDataSrc::close {}", self.id));
441        }
442        fn create_data_conn(&mut self) -> errs::Result<Box<MyDataConn>> {
443            if self.failure == Failure::FailToCreateDataConn {
444                self.logger
445                    .lock()
446                    .unwrap()
447                    .push(format!("MyDataSrc::create_data_conn {} failed", self.id));
448                return Err(errs::Err::new("eeee".to_string()));
449            }
450            {
451                self.logger
452                    .lock()
453                    .unwrap()
454                    .push(format!("MyDataSrc::create_data_conn {}", self.id));
455            }
456            let conn = MyDataConn::new(self.id, self.logger.clone(), self.failure);
457            Ok(Box::new(conn))
458        }
459    }
460
461    struct AnotherDataConn {}
462    impl DataConn for AnotherDataConn {
463        fn pre_commit(&mut self, _ag: &mut AsyncGroup) -> errs::Result<()> {
464            Ok(())
465        }
466        fn commit(&mut self, _ag: &mut AsyncGroup) -> errs::Result<()> {
467            Ok(())
468        }
469        fn is_committed(&self) -> bool {
470            false
471        }
472        fn post_commit(&mut self, _ag: &mut AsyncGroup) -> errs::Result<()> {
473            Ok(())
474        }
475        fn rollback(&mut self, _ag: &mut AsyncGroup) -> errs::Result<()> {
476            Ok(())
477        }
478        fn on_txn_failure(&mut self, _ag: &mut AsyncGroup, _reports: &[TxnFailureReport]) {}
479        fn close(&mut self) {}
480    }
481
482    #[test]
483    fn test_new() {
484        let hub = DataHub::new();
485        assert!(hub.local_data_src_manager.vec_unready.is_empty());
486        assert!(hub.local_data_src_manager.vec_ready.is_empty());
487        assert!(hub.local_data_src_manager.local);
488        assert!(hub.data_src_map.is_empty());
489        assert!(hub.data_conn_manager.vec.is_empty());
490        assert!(hub.data_conn_manager.index_map.is_empty());
491        assert!(!hub.fixed);
492    }
493
494    #[test]
495    fn test_with_commit_order() {
496        let hub = DataHub::with_commit_order(&["bar", "qux", "foo"]);
497        assert!(hub.local_data_src_manager.vec_unready.is_empty());
498        assert!(hub.local_data_src_manager.vec_ready.is_empty());
499        assert!(hub.local_data_src_manager.local);
500        assert!(hub.data_src_map.is_empty());
501        assert_eq!(hub.data_conn_manager.vec.len(), 3);
502        assert_eq!(hub.data_conn_manager.index_map.len(), 3);
503        assert!(!hub.fixed);
504    }
505
506    #[test]
507    fn test_uses_and_ok() {
508        let logger = Arc::new(Mutex::new(Vec::<String>::new()));
509
510        let mut hub = DataHub::new();
511        hub.uses("foo", MyDataSrc::new(1, logger.clone(), Failure::None));
512        hub.uses("bar", MyDataSrc::new(2, logger.clone(), Failure::None));
513
514        assert_eq!(hub.local_data_src_manager.vec_unready.len(), 2);
515        assert!(hub.local_data_src_manager.vec_ready.is_empty());
516        assert!(hub.local_data_src_manager.local);
517        assert!(hub.data_src_map.is_empty());
518        assert_eq!(hub.data_conn_manager.vec.len(), 0);
519        assert_eq!(hub.data_conn_manager.index_map.len(), 0);
520        assert!(!hub.fixed);
521
522        assert!(hub.begin().is_ok());
523
524        assert_eq!(hub.local_data_src_manager.vec_unready.len(), 0);
525        assert_eq!(hub.local_data_src_manager.vec_ready.len(), 2);
526        assert!(hub.local_data_src_manager.local);
527        assert_eq!(hub.data_src_map.len(), 2);
528        assert_eq!(hub.data_conn_manager.vec.len(), 0);
529        assert_eq!(hub.data_conn_manager.index_map.len(), 0);
530        assert!(hub.fixed);
531    }
532
533    #[test]
534    fn test_uses_but_already_fixed() {
535        let logger = Arc::new(Mutex::new(Vec::<String>::new()));
536
537        let mut hub = DataHub::new();
538        hub.uses("foo", MyDataSrc::new(1, logger.clone(), Failure::None));
539
540        assert_eq!(hub.local_data_src_manager.vec_unready.len(), 1);
541        assert_eq!(hub.local_data_src_manager.vec_ready.len(), 0);
542        assert!(hub.local_data_src_manager.local);
543        assert_eq!(hub.data_src_map.len(), 0);
544        assert_eq!(hub.data_conn_manager.vec.len(), 0);
545        assert_eq!(hub.data_conn_manager.index_map.len(), 0);
546        assert!(!hub.fixed);
547
548        assert!(hub.begin().is_ok());
549
550        assert_eq!(hub.local_data_src_manager.vec_unready.len(), 0);
551        assert_eq!(hub.local_data_src_manager.vec_ready.len(), 1);
552        assert!(hub.local_data_src_manager.local);
553        assert_eq!(hub.data_src_map.len(), 1);
554        assert_eq!(hub.data_conn_manager.vec.len(), 0);
555        assert_eq!(hub.data_conn_manager.index_map.len(), 0);
556        assert!(hub.fixed);
557
558        hub.uses("bar", MyDataSrc::new(2, logger.clone(), Failure::None));
559
560        assert_eq!(hub.local_data_src_manager.vec_unready.len(), 0);
561        assert_eq!(hub.local_data_src_manager.vec_ready.len(), 1);
562        assert!(hub.local_data_src_manager.local);
563        assert_eq!(hub.data_src_map.len(), 1);
564        assert_eq!(hub.data_conn_manager.vec.len(), 0);
565        assert_eq!(hub.data_conn_manager.index_map.len(), 0);
566        assert!(hub.fixed);
567    }
568
569    #[test]
570    fn test_disuses_and_ok() {
571        let logger = Arc::new(Mutex::new(Vec::<String>::new()));
572
573        let mut hub = DataHub::new();
574        hub.uses("foo", MyDataSrc::new(1, logger.clone(), Failure::None));
575        hub.uses("bar", MyDataSrc::new(2, logger.clone(), Failure::None));
576
577        assert_eq!(hub.local_data_src_manager.vec_unready.len(), 2);
578        assert!(hub.local_data_src_manager.vec_ready.is_empty());
579        assert!(hub.local_data_src_manager.local);
580        assert!(hub.data_src_map.is_empty());
581        assert_eq!(hub.data_conn_manager.vec.len(), 0);
582        assert_eq!(hub.data_conn_manager.index_map.len(), 0);
583        assert!(!hub.fixed);
584
585        hub.disuses("foo");
586
587        assert_eq!(hub.local_data_src_manager.vec_unready.len(), 1);
588        assert!(hub.local_data_src_manager.vec_ready.is_empty());
589        assert!(hub.local_data_src_manager.local);
590        assert!(hub.data_src_map.is_empty());
591        assert_eq!(hub.data_conn_manager.vec.len(), 0);
592        assert_eq!(hub.data_conn_manager.index_map.len(), 0);
593        assert!(!hub.fixed);
594
595        hub.disuses("bar");
596
597        assert_eq!(hub.local_data_src_manager.vec_unready.len(), 0);
598        assert!(hub.local_data_src_manager.vec_ready.is_empty());
599        assert!(hub.local_data_src_manager.local);
600        assert!(hub.data_src_map.is_empty());
601        assert_eq!(hub.data_conn_manager.vec.len(), 0);
602        assert_eq!(hub.data_conn_manager.index_map.len(), 0);
603        assert!(!hub.fixed);
604    }
605
606    #[test]
607    fn test_disuses_and_fix() {
608        let logger = Arc::new(Mutex::new(Vec::<String>::new()));
609
610        let mut hub = DataHub::new();
611        hub.uses("foo", MyDataSrc::new(1, logger.clone(), Failure::None));
612        hub.uses("bar", MyDataSrc::new(2, logger.clone(), Failure::None));
613
614        assert_eq!(hub.local_data_src_manager.vec_unready.len(), 2);
615        assert!(hub.local_data_src_manager.vec_ready.is_empty());
616        assert!(hub.local_data_src_manager.local);
617        assert!(hub.data_src_map.is_empty());
618        assert_eq!(hub.data_conn_manager.vec.len(), 0);
619        assert_eq!(hub.data_conn_manager.index_map.len(), 0);
620        assert!(!hub.fixed);
621
622        hub.disuses("foo");
623
624        assert_eq!(hub.local_data_src_manager.vec_unready.len(), 1);
625        assert!(hub.local_data_src_manager.vec_ready.is_empty());
626        assert!(hub.local_data_src_manager.local);
627        assert!(hub.data_src_map.is_empty());
628        assert_eq!(hub.data_conn_manager.vec.len(), 0);
629        assert_eq!(hub.data_conn_manager.index_map.len(), 0);
630        assert!(!hub.fixed);
631
632        hub.disuses("bar");
633
634        assert_eq!(hub.local_data_src_manager.vec_unready.len(), 0);
635        assert!(hub.local_data_src_manager.vec_ready.is_empty());
636        assert!(hub.local_data_src_manager.local);
637        assert!(hub.data_src_map.is_empty());
638        assert_eq!(hub.data_conn_manager.vec.len(), 0);
639        assert_eq!(hub.data_conn_manager.index_map.len(), 0);
640        assert!(!hub.fixed);
641
642        hub.uses("foo", MyDataSrc::new(1, logger.clone(), Failure::None));
643        hub.uses("bar", MyDataSrc::new(2, logger.clone(), Failure::None));
644
645        assert!(hub.begin().is_ok());
646
647        assert!(hub.local_data_src_manager.vec_unready.is_empty());
648        assert_eq!(hub.local_data_src_manager.vec_ready.len(), 2);
649        assert!(hub.local_data_src_manager.local);
650        assert_eq!(hub.data_src_map.len(), 2);
651        assert_eq!(hub.data_conn_manager.vec.len(), 0);
652        assert_eq!(hub.data_conn_manager.index_map.len(), 0);
653        assert!(hub.fixed);
654
655        hub.uses("baz", MyDataSrc::new(3, logger.clone(), Failure::None));
656
657        assert!(hub.local_data_src_manager.vec_unready.is_empty());
658        assert_eq!(hub.local_data_src_manager.vec_ready.len(), 2);
659        assert!(hub.local_data_src_manager.local);
660        assert_eq!(hub.data_src_map.len(), 2);
661        assert_eq!(hub.data_conn_manager.vec.len(), 0);
662        assert_eq!(hub.data_conn_manager.index_map.len(), 0);
663        assert!(hub.fixed);
664
665        hub.disuses("bar");
666
667        assert!(hub.local_data_src_manager.vec_unready.is_empty());
668        assert_eq!(hub.local_data_src_manager.vec_ready.len(), 2);
669        assert!(hub.local_data_src_manager.local);
670        assert_eq!(hub.data_src_map.len(), 2);
671        assert_eq!(hub.data_conn_manager.vec.len(), 0);
672        assert_eq!(hub.data_conn_manager.index_map.len(), 0);
673        assert!(hub.fixed);
674
675        hub.end();
676
677        assert!(hub.local_data_src_manager.vec_unready.is_empty());
678        assert_eq!(hub.local_data_src_manager.vec_ready.len(), 2);
679        assert!(hub.local_data_src_manager.local);
680        assert_eq!(hub.data_src_map.len(), 2);
681        assert_eq!(hub.data_conn_manager.vec.len(), 0);
682        assert_eq!(hub.data_conn_manager.index_map.len(), 0);
683        assert!(!hub.fixed);
684
685        hub.disuses("bar");
686
687        assert!(hub.local_data_src_manager.vec_unready.is_empty());
688        assert_eq!(hub.local_data_src_manager.vec_ready.len(), 1);
689        assert!(hub.local_data_src_manager.local);
690        assert_eq!(hub.data_src_map.len(), 1);
691        assert_eq!(hub.data_conn_manager.vec.len(), 0);
692        assert_eq!(hub.data_conn_manager.index_map.len(), 0);
693        assert!(!hub.fixed);
694
695        hub.disuses("foo");
696
697        assert!(hub.local_data_src_manager.vec_unready.is_empty());
698        assert!(hub.local_data_src_manager.vec_ready.is_empty());
699        assert!(hub.local_data_src_manager.local);
700        assert_eq!(hub.data_src_map.len(), 0);
701        assert_eq!(hub.data_conn_manager.vec.len(), 0);
702        assert_eq!(hub.data_conn_manager.index_map.len(), 0);
703        assert!(!hub.fixed);
704    }
705
706    #[test]
707    fn test_begin_if_empty() {
708        let mut hub = DataHub::new();
709        assert!(hub.begin().is_ok());
710
711        assert!(hub.local_data_src_manager.vec_unready.is_empty());
712        assert!(hub.local_data_src_manager.vec_ready.is_empty());
713        assert!(hub.local_data_src_manager.local);
714        assert_eq!(hub.data_src_map.len(), 0);
715        assert_eq!(hub.data_conn_manager.vec.len(), 0);
716        assert_eq!(hub.data_conn_manager.index_map.len(), 0);
717        assert!(hub.fixed);
718
719        hub.end();
720
721        assert!(hub.local_data_src_manager.vec_unready.is_empty());
722        assert!(hub.local_data_src_manager.vec_ready.is_empty());
723        assert!(hub.local_data_src_manager.local);
724        assert_eq!(hub.data_src_map.len(), 0);
725        assert_eq!(hub.data_conn_manager.vec.len(), 0);
726        assert_eq!(hub.data_conn_manager.index_map.len(), 0);
727        assert!(!hub.fixed);
728    }
729
730    #[test]
731    fn test_begin_and_ok() {
732        let logger = Arc::new(Mutex::new(Vec::<String>::new()));
733
734        {
735            let mut hub = DataHub::new();
736
737            hub.uses("foo", MyDataSrc::new(1, logger.clone(), Failure::None));
738            hub.uses("bar", MyDataSrc::new(2, logger.clone(), Failure::None));
739
740            assert_eq!(hub.local_data_src_manager.vec_unready.len(), 2);
741            assert_eq!(hub.local_data_src_manager.vec_ready.len(), 0);
742            assert_eq!(hub.local_data_src_manager.local, true);
743            assert_eq!(hub.data_src_map.len(), 0);
744            assert_eq!(hub.data_conn_manager.vec.len(), 0);
745            assert_eq!(hub.data_conn_manager.index_map.len(), 0);
746            assert_eq!(hub.fixed, false);
747
748            assert_eq!(hub.begin().is_ok(), true);
749
750            assert_eq!(hub.local_data_src_manager.vec_unready.len(), 0);
751            assert_eq!(hub.local_data_src_manager.vec_ready.len(), 2);
752            assert_eq!(hub.local_data_src_manager.local, true);
753            assert_eq!(hub.data_src_map.len(), 2);
754            assert_eq!(hub.data_conn_manager.vec.len(), 0);
755            assert_eq!(hub.data_conn_manager.index_map.len(), 0);
756            assert_eq!(hub.fixed, true);
757
758            hub.end();
759
760            assert_eq!(hub.local_data_src_manager.vec_unready.len(), 0);
761            assert_eq!(hub.local_data_src_manager.vec_ready.len(), 2);
762            assert_eq!(hub.local_data_src_manager.local, true);
763            assert_eq!(hub.data_src_map.len(), 2);
764            assert_eq!(hub.data_conn_manager.vec.len(), 0);
765            assert_eq!(hub.data_conn_manager.index_map.len(), 0);
766            assert_eq!(hub.fixed, false);
767        }
768
769        assert_eq!(
770            *logger.lock().unwrap(),
771            &[
772                "MyDataSrc::new 1",
773                "MyDataSrc::new 2",
774                "MyDataSrc::setup 1",
775                "MyDataSrc::setup 2",
776                "MyDataSrc::close 2",
777                "MyDataSrc::drop 2",
778                "MyDataSrc::close 1",
779                "MyDataSrc::drop 1",
780            ]
781        );
782    }
783
784    #[test]
785    fn test_begin_but_failed() {
786        let logger = Arc::new(Mutex::new(Vec::<String>::new()));
787
788        {
789            let mut hub = DataHub::new();
790
791            hub.uses("foo", MyDataSrc::new(1, logger.clone(), Failure::None));
792            hub.uses(
793                "bar",
794                MyDataSrc::new(2, logger.clone(), Failure::FailToSetup),
795            );
796            hub.uses("baz", MyDataSrc::new(3, logger.clone(), Failure::None));
797
798            assert_eq!(hub.local_data_src_manager.vec_unready.len(), 3);
799            assert_eq!(hub.local_data_src_manager.vec_ready.len(), 0);
800            assert_eq!(hub.local_data_src_manager.local, true);
801            assert_eq!(hub.data_src_map.len(), 0);
802            assert_eq!(hub.data_conn_manager.vec.len(), 0);
803            assert_eq!(hub.data_conn_manager.index_map.len(), 0);
804            assert_eq!(hub.fixed, false);
805
806            if let Err(err) = hub.begin() {
807                match err.reason::<DataHubError>() {
808                    Ok(DataHubError::FailToSetupLocalDataSrcs { errors }) => {
809                        assert_eq!(errors.len(), 1);
810                        assert_eq!(errors[0].index, 1);
811                        assert_eq!(errors[0].name, "bar".into());
812                        assert_eq!(errors[0].err.reason::<String>().unwrap(), "setup error");
813                    }
814                    _ => panic!(),
815                }
816            } else {
817                panic!();
818            }
819
820            hub.end();
821        }
822
823        assert_eq!(
824            *logger.lock().unwrap(),
825            &[
826                "MyDataSrc::new 1",
827                "MyDataSrc::new 2",
828                "MyDataSrc::new 3",
829                "MyDataSrc::setup 1",
830                "MyDataSrc::setup 2 failed",
831                "MyDataSrc::close 1",
832                "MyDataSrc::drop 3",
833                "MyDataSrc::drop 2",
834                "MyDataSrc::drop 1",
835            ]
836        );
837    }
838
839    #[test]
840    fn test_run_and_ok() {
841        let logger = Arc::new(Mutex::new(Vec::<String>::new()));
842        {
843            let mut hub = DataHub::new();
844
845            hub.uses("foo", MyDataSrc::new(1, logger.clone(), Failure::None));
846            hub.uses("bar", MyDataSrc::new(2, logger.clone(), Failure::None));
847
848            let logger_clone = logger.clone();
849            assert!(hub
850                .run(move |_data| {
851                    logger_clone
852                        .lock()
853                        .unwrap()
854                        .push("execute logic".to_string());
855                    Ok(())
856                })
857                .is_ok());
858        }
859
860        assert_eq!(
861            *logger.lock().unwrap(),
862            &[
863                "MyDataSrc::new 1",
864                "MyDataSrc::new 2",
865                "MyDataSrc::setup 1",
866                "MyDataSrc::setup 2",
867                "execute logic",
868                "MyDataSrc::close 2",
869                "MyDataSrc::drop 2",
870                "MyDataSrc::close 1",
871                "MyDataSrc::drop 1",
872            ]
873        );
874    }
875
876    #[test]
877    fn test_run_but_failed_to_run_logic() {
878        let logger = Arc::new(Mutex::new(Vec::<String>::new()));
879        {
880            let mut hub = DataHub::new();
881
882            hub.uses("foo", MyDataSrc::new(1, logger.clone(), Failure::None));
883            hub.uses("bar", MyDataSrc::new(2, logger.clone(), Failure::None));
884
885            let logger_clone = logger.clone();
886            if let Err(err) = hub.run(move |_data| {
887                logger_clone
888                    .lock()
889                    .unwrap()
890                    .push("execute logic but fail".to_string());
891                Err(errs::Err::new("logic error".to_string()))
892            }) {
893                match err.reason::<String>() {
894                    Ok(s) => assert_eq!(s, "logic error"),
895                    _ => panic!(),
896                }
897            } else {
898                panic!();
899            }
900        }
901
902        assert_eq!(
903            *logger.lock().unwrap(),
904            &[
905                "MyDataSrc::new 1",
906                "MyDataSrc::new 2",
907                "MyDataSrc::setup 1",
908                "MyDataSrc::setup 2",
909                "execute logic but fail",
910                "MyDataSrc::close 2",
911                "MyDataSrc::drop 2",
912                "MyDataSrc::close 1",
913                "MyDataSrc::drop 1",
914            ]
915        );
916    }
917
918    #[test]
919    fn test_txn_and_no_data_access_and_ok() {
920        let logger = Arc::new(Mutex::new(Vec::<String>::new()));
921        {
922            let mut hub = DataHub::new();
923
924            hub.uses("foo", MyDataSrc::new(1, logger.clone(), Failure::None));
925            hub.uses("bar", MyDataSrc::new(2, logger.clone(), Failure::None));
926
927            let logger_clone = logger.clone();
928            assert!(hub
929                .txn(move |_data| {
930                    logger_clone
931                        .lock()
932                        .unwrap()
933                        .push("execute logic".to_string());
934                    Ok(())
935                })
936                .is_ok());
937        }
938
939        assert_eq!(
940            *logger.lock().unwrap(),
941            &[
942                "MyDataSrc::new 1",
943                "MyDataSrc::new 2",
944                "MyDataSrc::setup 1",
945                "MyDataSrc::setup 2",
946                "execute logic",
947                "MyDataSrc::close 2",
948                "MyDataSrc::drop 2",
949                "MyDataSrc::close 1",
950                "MyDataSrc::drop 1",
951            ]
952        );
953    }
954
955    #[test]
956    fn test_txn_and_has_data_access_and_ok() {
957        let logger = Arc::new(Mutex::new(Vec::<String>::new()));
958        {
959            let mut hub = DataHub::new();
960
961            hub.uses("foo", MyDataSrc::new(1, logger.clone(), Failure::None));
962            hub.uses("bar", MyDataSrc::new(2, logger.clone(), Failure::None));
963
964            let logger_clone = logger.clone();
965            hub.txn(move |data| {
966                logger_clone
967                    .lock()
968                    .unwrap()
969                    .push("execute logic".to_string());
970                let _conn1 = data.get_data_conn::<MyDataConn>("foo")?;
971                let _conn2 = data.get_data_conn::<MyDataConn>("bar")?;
972                Ok(())
973            })
974            .unwrap();
975        }
976
977        assert_eq!(
978            *logger.lock().unwrap(),
979            &[
980                "MyDataSrc::new 1",
981                "MyDataSrc::new 2",
982                "MyDataSrc::setup 1",
983                "MyDataSrc::setup 2",
984                "execute logic",
985                "MyDataSrc::create_data_conn 1",
986                "MyDataConn::new 1",
987                "MyDataSrc::create_data_conn 2",
988                "MyDataConn::new 2",
989                "MyDataConn::pre_commit 1",
990                "MyDataConn::pre_commit 2",
991                "MyDataConn::commit 1",
992                "MyDataConn::commit 2",
993                "MyDataConn::post_commit 1",
994                "MyDataConn::post_commit 2",
995                "MyDataConn::close 2",
996                "MyDataConn::drop 2",
997                "MyDataConn::close 1",
998                "MyDataConn::drop 1",
999                "MyDataSrc::close 2",
1000                "MyDataSrc::drop 2",
1001                "MyDataSrc::close 1",
1002                "MyDataSrc::drop 1",
1003            ]
1004        );
1005    }
1006
1007    #[test]
1008    fn test_txn_but_failed_to_run_logic() {
1009        let logger = Arc::new(Mutex::new(Vec::<String>::new()));
1010        {
1011            let mut hub = DataHub::new();
1012
1013            hub.uses("foo", MyDataSrc::new(1, logger.clone(), Failure::None));
1014            hub.uses("bar", MyDataSrc::new(2, logger.clone(), Failure::None));
1015
1016            let logger_clone = logger.clone();
1017            if let Err(e) = hub.txn(move |data| {
1018                logger_clone
1019                    .lock()
1020                    .unwrap()
1021                    .push("execute logic".to_string());
1022                let _conn1 = data.get_data_conn::<MyDataConn>("foo")?;
1023                let _conn2 = data.get_data_conn::<MyDataConn>("bar")?;
1024                Err(errs::Err::new("logic error"))
1025            }) {
1026                match e.reason::<&str>() {
1027                    Ok(s) => assert_eq!(s, &"logic error"),
1028                    _ => panic!(),
1029                }
1030            }
1031        }
1032
1033        assert_eq!(
1034            *logger.lock().unwrap(),
1035            &[
1036                "MyDataSrc::new 1",
1037                "MyDataSrc::new 2",
1038                "MyDataSrc::setup 1",
1039                "MyDataSrc::setup 2",
1040                "execute logic",
1041                "MyDataSrc::create_data_conn 1",
1042                "MyDataConn::new 1",
1043                "MyDataSrc::create_data_conn 2",
1044                "MyDataConn::new 2",
1045                "MyDataConn::rollback 1",
1046                "MyDataConn::rollback 2",
1047                "MyDataConn::on_txn_failure 1",
1048                "MyDataConn::on_txn_failure 2",
1049                "MyDataConn::close 2",
1050                "MyDataConn::drop 2",
1051                "MyDataConn::close 1",
1052                "MyDataConn::drop 1",
1053                "MyDataSrc::close 2",
1054                "MyDataSrc::drop 2",
1055                "MyDataSrc::close 1",
1056                "MyDataSrc::drop 1",
1057            ]
1058        );
1059    }
1060
1061    #[test]
1062    fn test_txn_but_failed_to_pre_commit() {
1063        let logger = Arc::new(Mutex::new(Vec::<String>::new()));
1064        {
1065            let mut hub = DataHub::new();
1066
1067            hub.uses(
1068                "foo",
1069                MyDataSrc::new(1, logger.clone(), Failure::FailToPreCommit),
1070            );
1071            hub.uses(
1072                "bar",
1073                MyDataSrc::new(2, logger.clone(), Failure::FailToPreCommit),
1074            );
1075
1076            let logger_clone = logger.clone();
1077            if let Err(e) = hub.txn(move |data| {
1078                logger_clone
1079                    .lock()
1080                    .unwrap()
1081                    .push("execute logic".to_string());
1082                let _conn1 = data.get_data_conn::<MyDataConn>("foo")?;
1083                let _conn2 = data.get_data_conn::<MyDataConn>("bar")?;
1084                Ok(())
1085            }) {
1086                match e.reason::<DataConnError>() {
1087                    Ok(DataConnError::FailToPreCommitDataConn { errors }) => {
1088                        assert_eq!(errors.len(), 1);
1089                        assert_eq!(errors[0].index, 0);
1090                        assert_eq!(errors[0].name, "foo".into());
1091                        assert_eq!(errors[0].err.reason::<&str>().unwrap(), &"pre commit error");
1092                    }
1093                    _ => panic!(),
1094                }
1095            }
1096        }
1097
1098        assert_eq!(
1099            *logger.lock().unwrap(),
1100            &[
1101                "MyDataSrc::new 1",
1102                "MyDataSrc::new 2",
1103                "MyDataSrc::setup 1",
1104                "MyDataSrc::setup 2",
1105                "execute logic",
1106                "MyDataSrc::create_data_conn 1",
1107                "MyDataConn::new 1",
1108                "MyDataSrc::create_data_conn 2",
1109                "MyDataConn::new 2",
1110                "MyDataConn::pre_commit 1 failed",
1111                "MyDataConn::rollback 1",
1112                "MyDataConn::rollback 2",
1113                "MyDataConn::on_txn_failure 1",
1114                "MyDataConn::on_txn_failure 2",
1115                "MyDataConn::close 2",
1116                "MyDataConn::drop 2",
1117                "MyDataConn::close 1",
1118                "MyDataConn::drop 1",
1119                "MyDataSrc::close 2",
1120                "MyDataSrc::drop 2",
1121                "MyDataSrc::close 1",
1122                "MyDataSrc::drop 1",
1123            ]
1124        );
1125    }
1126
1127    #[test]
1128    fn test_txn_but_failed_to_commit() {
1129        let logger = Arc::new(Mutex::new(Vec::<String>::new()));
1130        {
1131            let mut hub = DataHub::new();
1132
1133            hub.uses(
1134                "foo",
1135                MyDataSrc::new(1, logger.clone(), Failure::FailToCommit),
1136            );
1137            hub.uses(
1138                "bar",
1139                MyDataSrc::new(2, logger.clone(), Failure::FailToCommit),
1140            );
1141
1142            let logger_clone = logger.clone();
1143            if let Err(e) = hub.txn(move |data| {
1144                logger_clone
1145                    .lock()
1146                    .unwrap()
1147                    .push("execute logic".to_string());
1148                let _conn1 = data.get_data_conn::<MyDataConn>("foo")?;
1149                let _conn2 = data.get_data_conn::<MyDataConn>("bar")?;
1150                Ok(())
1151            }) {
1152                match e.reason::<DataConnError>() {
1153                    Ok(DataConnError::FailToCommitDataConn { errors }) => {
1154                        assert_eq!(errors.len(), 1);
1155                        assert_eq!(errors[0].index, 0);
1156                        assert_eq!(errors[0].name, "foo".into());
1157                        assert_eq!(errors[0].err.reason::<&str>().unwrap(), &"commit error");
1158                    }
1159                    _ => panic!(),
1160                }
1161            }
1162        }
1163
1164        assert_eq!(
1165            *logger.lock().unwrap(),
1166            &[
1167                "MyDataSrc::new 1",
1168                "MyDataSrc::new 2",
1169                "MyDataSrc::setup 1",
1170                "MyDataSrc::setup 2",
1171                "execute logic",
1172                "MyDataSrc::create_data_conn 1",
1173                "MyDataConn::new 1",
1174                "MyDataSrc::create_data_conn 2",
1175                "MyDataConn::new 2",
1176                "MyDataConn::pre_commit 1",
1177                "MyDataConn::pre_commit 2",
1178                "MyDataConn::commit 1 failed",
1179                "MyDataConn::rollback 1",
1180                "MyDataConn::rollback 2",
1181                "MyDataConn::on_txn_failure 1",
1182                "MyDataConn::on_txn_failure 2",
1183                "MyDataConn::close 2",
1184                "MyDataConn::drop 2",
1185                "MyDataConn::close 1",
1186                "MyDataConn::drop 1",
1187                "MyDataSrc::close 2",
1188                "MyDataSrc::drop 2",
1189                "MyDataSrc::close 1",
1190                "MyDataSrc::drop 1",
1191            ]
1192        );
1193    }
1194
1195    #[test]
1196    fn test_txn_but_failed_to_post_commit() {
1197        let logger = Arc::new(Mutex::new(Vec::<String>::new()));
1198        {
1199            let mut hub = DataHub::new();
1200
1201            hub.uses(
1202                "foo",
1203                MyDataSrc::new(1, logger.clone(), Failure::FailToPostCommit),
1204            );
1205            hub.uses(
1206                "bar",
1207                MyDataSrc::new(2, logger.clone(), Failure::FailToPostCommit),
1208            );
1209
1210            let logger_clone = logger.clone();
1211            if let Err(e) = hub.txn(move |data| {
1212                logger_clone
1213                    .lock()
1214                    .unwrap()
1215                    .push("execute logic".to_string());
1216                let _conn1 = data.get_data_conn::<MyDataConn>("foo")?;
1217                let _conn2 = data.get_data_conn::<MyDataConn>("bar")?;
1218                Ok(())
1219            }) {
1220                match e.reason::<DataConnError>() {
1221                    Ok(DataConnError::FailToPostCommitDataConn { errors }) => {
1222                        assert_eq!(errors.len(), 2);
1223                        assert_eq!(errors[0].index, 0);
1224                        assert_eq!(errors[0].name, "foo".into());
1225                        assert_eq!(
1226                            errors[0].err.reason::<&str>().unwrap(),
1227                            &"post commit error"
1228                        );
1229                        assert_eq!(errors[1].index, 1);
1230                        assert_eq!(errors[1].name, "bar".into());
1231                        assert_eq!(
1232                            errors[1].err.reason::<&str>().unwrap(),
1233                            &"post commit error"
1234                        );
1235                    }
1236                    _ => panic!(),
1237                }
1238            }
1239        }
1240
1241        assert_eq!(
1242            *logger.lock().unwrap(),
1243            &[
1244                "MyDataSrc::new 1",
1245                "MyDataSrc::new 2",
1246                "MyDataSrc::setup 1",
1247                "MyDataSrc::setup 2",
1248                "execute logic",
1249                "MyDataSrc::create_data_conn 1",
1250                "MyDataConn::new 1",
1251                "MyDataSrc::create_data_conn 2",
1252                "MyDataConn::new 2",
1253                "MyDataConn::pre_commit 1",
1254                "MyDataConn::pre_commit 2",
1255                "MyDataConn::commit 1",
1256                "MyDataConn::commit 2",
1257                "MyDataConn::post_commit 1 failed",
1258                "MyDataConn::post_commit 2 failed",
1259                "MyDataConn::on_txn_failure 1",
1260                "MyDataConn::on_txn_failure 2",
1261                "MyDataConn::close 2",
1262                "MyDataConn::drop 2",
1263                "MyDataConn::close 1",
1264                "MyDataConn::drop 1",
1265                "MyDataSrc::close 2",
1266                "MyDataSrc::drop 2",
1267                "MyDataSrc::close 1",
1268                "MyDataSrc::drop 1",
1269            ]
1270        );
1271    }
1272
1273    #[test]
1274    fn test_txn_but_failed_to_rollback() {
1275        let logger = Arc::new(Mutex::new(Vec::<String>::new()));
1276        {
1277            let mut hub = DataHub::new();
1278
1279            hub.uses(
1280                "foo",
1281                MyDataSrc::new(1, logger.clone(), Failure::FailToRollback),
1282            );
1283            hub.uses(
1284                "bar",
1285                MyDataSrc::new(2, logger.clone(), Failure::FailToRollback),
1286            );
1287
1288            let logger_clone = logger.clone();
1289            if let Err(e) = hub.txn(move |data| {
1290                logger_clone
1291                    .lock()
1292                    .unwrap()
1293                    .push("execute logic".to_string());
1294                let _conn1 = data.get_data_conn::<MyDataConn>("foo")?;
1295                let _conn2 = data.get_data_conn::<MyDataConn>("bar")?;
1296                Err(errs::Err::new("logic error"))
1297            }) {
1298                match e.reason::<&str>() {
1299                    Ok(s) => assert_eq!(s, &"logic error"),
1300                    _ => panic!(),
1301                }
1302            }
1303        }
1304
1305        assert_eq!(
1306            *logger.lock().unwrap(),
1307            &[
1308                "MyDataSrc::new 1",
1309                "MyDataSrc::new 2",
1310                "MyDataSrc::setup 1",
1311                "MyDataSrc::setup 2",
1312                "execute logic",
1313                "MyDataSrc::create_data_conn 1",
1314                "MyDataConn::new 1",
1315                "MyDataSrc::create_data_conn 2",
1316                "MyDataConn::new 2",
1317                "MyDataConn::rollback 1 failed",
1318                "MyDataConn::rollback 2 failed",
1319                "MyDataConn::on_txn_failure 1",
1320                "MyDataConn::on_txn_failure 2",
1321                "MyDataConn::close 2",
1322                "MyDataConn::drop 2",
1323                "MyDataConn::close 1",
1324                "MyDataConn::drop 1",
1325                "MyDataSrc::close 2",
1326                "MyDataSrc::drop 2",
1327                "MyDataSrc::close 1",
1328                "MyDataSrc::drop 1",
1329            ]
1330        );
1331    }
1332
1333    #[test]
1334    fn test_txn_with_commit_order() {
1335        let logger = Arc::new(Mutex::new(Vec::<String>::new()));
1336        {
1337            let mut hub = DataHub::with_commit_order(&["bar", "foo"]);
1338
1339            hub.uses("foo", MyDataSrc::new(1, logger.clone(), Failure::None));
1340            hub.uses("bar", MyDataSrc::new(2, logger.clone(), Failure::None));
1341
1342            let logger_clone = logger.clone();
1343
1344            if let Err(e) = hub.txn(move |data| {
1345                logger_clone
1346                    .lock()
1347                    .unwrap()
1348                    .push("execute logic".to_string());
1349                let _conn1 = data.get_data_conn::<MyDataConn>("foo")?;
1350                let _conn2 = data.get_data_conn::<MyDataConn>("bar")?;
1351                Ok(())
1352            }) {
1353                match e.reason::<&str>() {
1354                    Ok(s) => assert_eq!(s, &"logic error"),
1355                    _ => panic!(),
1356                }
1357            }
1358        }
1359
1360        assert_eq!(
1361            *logger.lock().unwrap(),
1362            &[
1363                "MyDataSrc::new 1",
1364                "MyDataSrc::new 2",
1365                "MyDataSrc::setup 1",
1366                "MyDataSrc::setup 2",
1367                "execute logic",
1368                "MyDataSrc::create_data_conn 1",
1369                "MyDataConn::new 1",
1370                "MyDataSrc::create_data_conn 2",
1371                "MyDataConn::new 2",
1372                "MyDataConn::pre_commit 2",
1373                "MyDataConn::pre_commit 1",
1374                "MyDataConn::commit 2",
1375                "MyDataConn::commit 1",
1376                "MyDataConn::post_commit 2",
1377                "MyDataConn::post_commit 1",
1378                "MyDataConn::close 1",
1379                "MyDataConn::drop 1",
1380                "MyDataConn::close 2",
1381                "MyDataConn::drop 2",
1382                "MyDataSrc::close 2",
1383                "MyDataSrc::drop 2",
1384                "MyDataSrc::close 1",
1385                "MyDataSrc::drop 1",
1386            ]
1387        );
1388    }
1389
1390    #[test]
1391    fn test_txn_but_fail_to_setup() {
1392        let logger = Arc::new(Mutex::new(Vec::<String>::new()));
1393        {
1394            let mut hub = DataHub::new();
1395
1396            hub.uses(
1397                "foo",
1398                MyDataSrc::new(1, logger.clone(), Failure::FailToSetup),
1399            );
1400
1401            let logger_clone = logger.clone();
1402
1403            if let Err(e) = hub.txn(move |_data| {
1404                logger_clone
1405                    .lock()
1406                    .unwrap()
1407                    .push("execute logic".to_string());
1408                Ok(())
1409            }) {
1410                match e.reason::<DataHubError>() {
1411                    Ok(DataHubError::FailToSetupLocalDataSrcs { errors }) => {
1412                        assert_eq!(errors.len(), 1);
1413                        assert_eq!(errors[0].index, 0);
1414                        assert_eq!(errors[0].name, "foo".into());
1415                        assert_eq!(errors[0].err.reason::<String>().unwrap(), "setup error");
1416                    }
1417                    _ => panic!(),
1418                }
1419            }
1420        }
1421
1422        assert_eq!(
1423            *logger.lock().unwrap(),
1424            &[
1425                "MyDataSrc::new 1",
1426                "MyDataSrc::setup 1 failed",
1427                "MyDataSrc::drop 1",
1428            ]
1429        );
1430    }
1431
1432    #[test]
1433    fn test_get_data_conn_cached() {
1434        let logger = Arc::new(Mutex::new(Vec::<String>::new()));
1435        {
1436            let mut hub = DataHub::new();
1437
1438            hub.uses("foo", MyDataSrc::new(1, logger.clone(), Failure::None));
1439
1440            let logger_clone = logger.clone();
1441
1442            if let Err(e) = hub.txn(move |data| {
1443                logger_clone
1444                    .lock()
1445                    .unwrap()
1446                    .push("execute logic".to_string());
1447                let _conn1 = data.get_data_conn::<MyDataConn>("foo")?;
1448                let _conn1 = data.get_data_conn::<MyDataConn>("foo")?;
1449                Ok(())
1450            }) {
1451                panic!("{:?}", e);
1452            }
1453        }
1454
1455        assert_eq!(
1456            *logger.lock().unwrap(),
1457            &[
1458                "MyDataSrc::new 1",
1459                "MyDataSrc::setup 1",
1460                "execute logic",
1461                "MyDataSrc::create_data_conn 1",
1462                "MyDataConn::new 1",
1463                "MyDataConn::pre_commit 1",
1464                "MyDataConn::commit 1",
1465                "MyDataConn::post_commit 1",
1466                "MyDataConn::close 1",
1467                "MyDataConn::drop 1",
1468                "MyDataSrc::close 1",
1469                "MyDataSrc::drop 1",
1470            ]
1471        );
1472    }
1473
1474    #[test]
1475    fn test_get_data_conn_and_no_data_src_to_create_data_conn() {
1476        let logger = Arc::new(Mutex::new(Vec::<String>::new()));
1477        {
1478            let mut hub = DataHub::new();
1479
1480            let logger_clone = logger.clone();
1481
1482            if let Err(e) = hub.txn(move |data| {
1483                logger_clone
1484                    .lock()
1485                    .unwrap()
1486                    .push("execute logic".to_string());
1487                let _conn1 = data.get_data_conn::<MyDataConn>("foo")?;
1488                Ok(())
1489            }) {
1490                match e.reason::<DataHubError>() {
1491                    Ok(DataHubError::NoDataSrcToCreateDataConn {
1492                        name,
1493                        data_conn_type,
1494                    }) => {
1495                        assert_eq!(name.as_ref(), "foo");
1496                        assert_eq!(
1497                            data_conn_type,
1498                            &"sabi::data_hub::tests_of_data_hub::MyDataConn"
1499                        );
1500                    }
1501                    _ => panic!(),
1502                }
1503            }
1504        }
1505
1506        assert_eq!(*logger.lock().unwrap(), &["execute logic",]);
1507    }
1508
1509    #[test]
1510    fn test_get_data_conn_and_failed_to_creata_data_conn() {
1511        let logger = Arc::new(Mutex::new(Vec::<String>::new()));
1512        {
1513            let mut hub = DataHub::new();
1514
1515            hub.uses(
1516                "foo",
1517                MyDataSrc::new(1, logger.clone(), Failure::FailToCreateDataConn),
1518            );
1519
1520            let logger_clone = logger.clone();
1521
1522            if let Err(e) = hub.txn(move |data| {
1523                logger_clone
1524                    .lock()
1525                    .unwrap()
1526                    .push("execute logic".to_string());
1527                let _conn1 = data.get_data_conn::<MyDataConn>("foo")?;
1528                Ok(())
1529            }) {
1530                match e.reason::<DataSrcError>() {
1531                    Ok(DataSrcError::FailToCreateDataConn {
1532                        name,
1533                        data_conn_type,
1534                    }) => {
1535                        assert_eq!(name.as_ref(), "foo");
1536                        assert_eq!(
1537                            data_conn_type,
1538                            &"sabi::data_hub::tests_of_data_hub::MyDataConn"
1539                        );
1540                    }
1541                    _ => panic!(),
1542                }
1543            }
1544        }
1545
1546        assert_eq!(
1547            *logger.lock().unwrap(),
1548            &[
1549                "MyDataSrc::new 1",
1550                "MyDataSrc::setup 1",
1551                "execute logic",
1552                "MyDataSrc::create_data_conn 1 failed",
1553                "MyDataSrc::close 1",
1554                "MyDataSrc::drop 1",
1555            ]
1556        );
1557    }
1558
1559    #[test]
1560    fn test_get_data_conn_and_failed_to_cast_data_conn() {
1561        let logger = Arc::new(Mutex::new(Vec::<String>::new()));
1562        {
1563            let mut hub = DataHub::new();
1564
1565            hub.uses("foo", MyDataSrc::new(1, logger.clone(), Failure::None));
1566
1567            let logger_clone = logger.clone();
1568
1569            if let Err(e) = hub.txn(move |data| {
1570                logger_clone
1571                    .lock()
1572                    .unwrap()
1573                    .push("execute logic".to_string());
1574                if let Err(e) = data.get_data_conn::<AnotherDataConn>("foo") {
1575                    match e.reason::<DataSrcError>() {
1576                        Ok(DataSrcError::FailToCastDataConn { name, target_type }) => {
1577                            assert_eq!(name.as_ref(), "foo");
1578                            assert_eq!(
1579                                target_type,
1580                                &"sabi::data_hub::tests_of_data_hub::AnotherDataConn"
1581                            );
1582                        }
1583                        _ => panic!("{e:?}"),
1584                    }
1585                } else {
1586                    panic!();
1587                }
1588
1589                let _conn1 = data.get_data_conn::<MyDataConn>("foo")?;
1590
1591                if let Err(e) = data.get_data_conn::<AnotherDataConn>("foo") {
1592                    match e.reason::<DataConnError>() {
1593                        Ok(DataConnError::FailToCastDataConn { name, target_type }) => {
1594                            assert_eq!(name.as_ref(), "foo");
1595                            assert_eq!(
1596                                target_type,
1597                                &"sabi::data_hub::tests_of_data_hub::AnotherDataConn"
1598                            );
1599                            Err(e)
1600                        }
1601                        _ => panic!("{e:?}"),
1602                    }
1603                } else {
1604                    panic!();
1605                }
1606            }) {
1607                match e.reason::<DataConnError>() {
1608                    Ok(DataConnError::FailToCastDataConn { name, target_type }) => {
1609                        assert_eq!(name.as_ref(), "foo");
1610                        assert_eq!(
1611                            target_type,
1612                            &"sabi::data_hub::tests_of_data_hub::AnotherDataConn"
1613                        );
1614                    }
1615                    _ => panic!(),
1616                }
1617            }
1618        }
1619
1620        assert_eq!(
1621            *logger.lock().unwrap(),
1622            &[
1623                "MyDataSrc::new 1",
1624                "MyDataSrc::setup 1",
1625                "execute logic",
1626                "MyDataSrc::create_data_conn 1",
1627                "MyDataConn::new 1",
1628                "MyDataConn::rollback 1",
1629                "MyDataConn::on_txn_failure 1",
1630                "MyDataConn::close 1",
1631                "MyDataConn::drop 1",
1632                "MyDataSrc::close 1",
1633                "MyDataSrc::drop 1",
1634            ]
1635        );
1636    }
1637
1638    #[test]
1639    fn data_hub_implements_send_trait() {
1640        let mut data = DataHub::new();
1641        let handle = std::thread::spawn(move || {
1642            data.run(|_data| Ok(())).unwrap();
1643        });
1644
1645        handle.join().unwrap();
1646    }
1647}
sabi/data_hub.rs

sabi/
data_hub.rs
