sabi/tokio/
data_acc.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::{DataAcc, DataConn, DataHub};
6
7impl DataAcc for DataHub {
8    /// Retrieves a data connection of a specific type from the `DataHub`.
9    ///
10    /// This asynchronous method attempts to get a data connection identified by `name`.
11    /// The connection type `C` must implement the `DataConn` trait and have a `'static` lifetime.
12    ///
13    /// # Parameters
14    ///
15    /// * `name` - An identifier for the data connection to retrieve.
16    ///
17    /// # Type Parameters
18    ///
19    /// * `C` - The expected type of the data connection, which must implement `DataConn`.
20    ///
21    /// # Returns
22    ///
23    /// A `Result` which is `Ok` containing a mutable reference to the data connection
24    /// if found and castable to type `C`, or an `Err` if the connection is not found
25    /// or cannot be cast.
26    async fn get_data_conn_async<C>(&mut self, name: &str) -> errs::Result<&mut C>
27    where
28        C: DataConn + 'static,
29    {
30        DataHub::get_data_conn_async(self, name).await
31    }
32}
33
34#[cfg_attr(coverage_nightly, coverage(off))]
35#[cfg(test)]
36mod tests_of_data_acc {
37    use super::super::{logic, AsyncGroup, DataSrc};
38    use super::*;
39    use crate::TxnFailureReport;
40    use std::sync::{Arc, Mutex};
41
42    struct FooDataConn {
43        id: i8,
44        text: String,
45        committed: bool,
46        logger: Arc<Mutex<Vec<String>>>,
47    }
48    impl FooDataConn {
49        fn new(id: i8, s: &str, logger: Arc<Mutex<Vec<String>>>) -> Self {
50            {
51                let mut logger = logger.lock().unwrap();
52                logger.push(format!("FooDataConn::new {}", id));
53            }
54            Self {
55                id,
56                text: s.to_string(),
57                logger,
58                committed: false,
59            }
60        }
61        async fn get_text_async(&self) -> String {
62            let mut logger = self.logger.lock().unwrap();
63            logger.push(format!("FooDataConn::get_text_async {}", self.id));
64            self.text.clone()
65        }
66    }
67    impl Drop for FooDataConn {
68        fn drop(&mut self) {
69            let mut logger = self.logger.lock().unwrap();
70            logger.push(format!("FooDataConn::drop {}", self.id));
71        }
72    }
73    impl DataConn for FooDataConn {
74        async fn commit_async(&mut self, _ag: &mut AsyncGroup) -> errs::Result<()> {
75            self.committed = true;
76            let mut logger = self.logger.lock().unwrap();
77            logger.push(format!("FooDataConn::commit_async {}", self.id));
78            Ok(())
79        }
80        async fn pre_commit_async(&mut self, _ag: &mut AsyncGroup) -> errs::Result<()> {
81            let mut logger = self.logger.lock().unwrap();
82            logger.push(format!("FooDataConn::pre_commit_async {}", self.id));
83            Ok(())
84        }
85        async fn post_commit_async(&mut self, _ag: &mut AsyncGroup) -> errs::Result<()> {
86            let mut logger = self.logger.lock().unwrap();
87            logger.push(format!("FooDataConn::post_commit_async {}", self.id));
88            Ok(())
89        }
90        fn is_committed(&self) -> bool {
91            self.committed
92        }
93        async fn rollback_async(&mut self, _ag: &mut AsyncGroup) -> errs::Result<()> {
94            let mut logger = self.logger.lock().unwrap();
95            logger.push(format!("FooDataConn::rollback_async {}", self.id));
96            Ok(())
97        }
98        async fn on_txn_failure_async(
99            &mut self,
100            _ag: &mut AsyncGroup,
101            _reports: Arc<[TxnFailureReport]>,
102        ) {
103            let mut logger = self.logger.lock().unwrap();
104            logger.push(format!("FooDataConn::on_txn_failure_async {}", self.id));
105        }
106        fn close(&mut self) {
107            let mut logger = self.logger.lock().unwrap();
108            logger.push(format!("FooDataConn::close {}", self.id));
109        }
110    }
111
112    struct FooDataSrc {
113        id: i8,
114        logger: Arc<Mutex<Vec<String>>>,
115        fail: bool,
116        text: String,
117    }
118    impl FooDataSrc {
119        fn new(id: i8, s: &str, logger: Arc<Mutex<Vec<String>>>, fail: bool) -> Self {
120            {
121                let mut logger = logger.lock().unwrap();
122                logger.push(format!("FooDataSrc::new {}", id));
123            }
124            Self {
125                id,
126                logger,
127                fail,
128                text: s.to_string(),
129            }
130        }
131    }
132    impl Drop for FooDataSrc {
133        fn drop(&mut self) {
134            let mut logger = self.logger.lock().unwrap();
135            logger.push(format!("FooDataSrc::drop {}", self.id));
136        }
137    }
138    impl DataSrc<FooDataConn> for FooDataSrc {
139        async fn setup_async(&mut self, _ag: &mut AsyncGroup) -> errs::Result<()> {
140            if self.fail {
141                {
142                    let mut logger = self.logger.lock().unwrap();
143                    logger.push(format!("FooDataSrc::setup_async {} failed", self.id));
144                }
145                return Err(errs::Err::new("XXXX".to_string()));
146            }
147            {
148                let mut logger = self.logger.lock().unwrap();
149                logger.push(format!("FooDataSrc::setup_async {}", self.id));
150            }
151            Ok(())
152        }
153        fn close(&mut self) {
154            let mut logger = self.logger.lock().unwrap();
155            logger.push(format!("FooDataSrc::close {}", self.id));
156        }
157        async fn create_data_conn_async(&mut self) -> errs::Result<Box<FooDataConn>> {
158            {
159                let mut logger = self.logger.lock().unwrap();
160                logger.push(format!("FooDataSrc::create_data_src_async {}", self.id));
161            }
162            let conn = FooDataConn::new(self.id, &self.text, self.logger.clone());
163            Ok(Box::new(conn))
164        }
165    }
166
167    struct BarDataConn {
168        id: i8,
169        text: Option<String>,
170        ds_text: Arc<tokio::sync::Mutex<String>>,
171        committed: bool,
172        logger: Arc<Mutex<Vec<String>>>,
173    }
174    impl BarDataConn {
175        fn new(
176            id: i8,
177            ds_text: Arc<tokio::sync::Mutex<String>>,
178            logger: Arc<Mutex<Vec<String>>>,
179        ) -> Self {
180            {
181                let mut logger = logger.lock().unwrap();
182                logger.push(format!("BarDataConn::new {}", id));
183            }
184            Self {
185                id,
186                text: None,
187                ds_text,
188                logger,
189                committed: false,
190            }
191        }
192        async fn set_text_async(&mut self, s: &str) {
193            let mut logger = self.logger.lock().unwrap();
194            logger.push(format!("BarDataConn::set_text_async {}", self.id));
195            self.text = Some(s.to_string());
196        }
197    }
198    impl Drop for BarDataConn {
199        fn drop(&mut self) {
200            let mut logger = self.logger.lock().unwrap();
201            logger.push(format!("BarDataConn::drop {}", self.id));
202        }
203    }
204    impl DataConn for BarDataConn {
205        async fn commit_async(&mut self, _ag: &mut AsyncGroup) -> errs::Result<()> {
206            self.committed = true;
207            match &self.text {
208                Some(s) => {
209                    let mut guard = self.ds_text.lock().await;
210                    *guard = s.to_string();
211                }
212                None => {
213                    let mut guard = self.ds_text.lock().await;
214                    *guard = "".to_string();
215                }
216            }
217            self.logger
218                .lock()
219                .unwrap()
220                .push(format!("BarDataConn::commit_async {}", self.id));
221            Ok(())
222        }
223        async fn pre_commit_async(&mut self, _ag: &mut AsyncGroup) -> errs::Result<()> {
224            let mut logger = self.logger.lock().unwrap();
225            logger.push(format!("BarDataConn::pre_commit_async {}", self.id));
226            Ok(())
227        }
228        async fn post_commit_async(&mut self, _ag: &mut AsyncGroup) -> errs::Result<()> {
229            let mut logger = self.logger.lock().unwrap();
230            logger.push(format!("BarDataConn::post_commit_async {}", self.id));
231            Ok(())
232        }
233        fn is_committed(&self) -> bool {
234            self.committed
235        }
236        async fn rollback_async(&mut self, _ag: &mut AsyncGroup) -> errs::Result<()> {
237            let mut logger = self.logger.lock().unwrap();
238            logger.push(format!("BarDataConn::rollback_async {}", self.id));
239            Ok(())
240        }
241        async fn on_txn_failure_async(
242            &mut self,
243            _ag: &mut AsyncGroup,
244            _reports: Arc<[TxnFailureReport]>,
245        ) {
246            let mut logger = self.logger.lock().unwrap();
247            logger.push(format!("BarDataConn::on_txn_failure_async {}", self.id));
248        }
249        fn close(&mut self) {
250            let mut logger = self.logger.lock().unwrap();
251            logger.push(format!("BarDataConn.text = {}", self.text.clone().unwrap()));
252            logger.push(format!("BarDataConn::close {}", self.id));
253        }
254    }
255
256    struct BarDataSrc {
257        id: i8,
258        text: Arc<tokio::sync::Mutex<String>>,
259        logger: Arc<Mutex<Vec<String>>>,
260    }
261    impl BarDataSrc {
262        fn new(id: i8, logger: Arc<Mutex<Vec<String>>>) -> Self {
263            {
264                let mut logger = logger.lock().unwrap();
265                logger.push(format!("BarDataSrc::new {}", id));
266            }
267            Self {
268                id,
269                text: Arc::new(tokio::sync::Mutex::new(String::new())),
270                logger,
271            }
272        }
273    }
274    impl Drop for BarDataSrc {
275        fn drop(&mut self) {
276            let mut logger = self.logger.lock().unwrap();
277            logger.push(format!("BarDataSrc::drop {}", self.id));
278        }
279    }
280    impl DataSrc<BarDataConn> for BarDataSrc {
281        async fn setup_async(&mut self, _ag: &mut AsyncGroup) -> errs::Result<()> {
282            let mut logger = self.logger.lock().unwrap();
283            logger.push(format!("BarDataSrc::setup_async {}", self.id));
284            Ok(())
285        }
286        fn close(&mut self) {
287            let mut logger = self.logger.lock().unwrap();
288            if let Ok(guard) = self.text.try_lock() {
289                logger.push(format!("BarDataSrc.text = {}", *guard));
290            }
291            logger.push(format!("BarDataSrc::close {}", self.id));
292        }
293        async fn create_data_conn_async(&mut self) -> errs::Result<Box<BarDataConn>> {
294            {
295                let mut logger = self.logger.lock().unwrap();
296                logger.push(format!("BarDataSrc::create_data_src_async {}", self.id));
297            }
298            let conn = BarDataConn::new(self.id, self.text.clone(), self.logger.clone());
299            Ok(Box::new(conn))
300        }
301    }
302
303    mod test_run_async_method {
304        use super::*;
305        use override_macro::{overridable, override_with};
306
307        #[overridable(mod = test_run_async_method)]
308        trait SampleAsyncData {
309            async fn get_value_async(&mut self) -> errs::Result<String>;
310            async fn set_value_async(&mut self, v: &str) -> errs::Result<()>;
311        }
312
313        async fn sample_logic_async(data: &mut (impl SampleAsyncData + Send)) -> errs::Result<()> {
314            let v = data.get_value_async().await?;
315            let _ = data.set_value_async(&v).await;
316            let v = data.get_value_async().await?;
317            let _ = data.set_value_async(&v).await;
318            Ok(())
319        }
320
321        #[overridable(mod = test_run_async_method)]
322        trait FooDataAcc: DataAcc {
323            async fn get_value_async(&mut self) -> errs::Result<String> {
324                let conn = self.get_data_conn_async::<FooDataConn>("foo").await?;
325                Ok(conn.get_text_async().await)
326            }
327        }
328
329        impl FooDataAcc for DataHub {}
330
331        #[overridable(mod = test_run_async_method)]
332        trait BarDataAcc: DataAcc {
333            async fn set_value_async(&mut self, text: &str) -> errs::Result<()> {
334                let conn = self.get_data_conn_async::<BarDataConn>("bar").await?;
335                conn.set_text_async(text).await;
336                Ok(())
337            }
338        }
339
340        impl BarDataAcc for DataHub {}
341
342        #[override_with(test_run_async_method::FooDataAcc, test_run_async_method::BarDataAcc)]
343        impl SampleAsyncData for DataHub {}
344
345        #[tokio::test]
346        async fn test_logic() {
347            let logger = Arc::new(Mutex::new(Vec::new()));
348
349            {
350                let mut data = DataHub::new();
351
352                data.uses("foo", FooDataSrc::new(1, "hello", logger.clone(), false));
353                data.uses("bar", BarDataSrc::new(2, logger.clone()));
354
355                if let Err(_) = data.run_async(logic!(sample_logic_async)).await {
356                    panic!();
357                }
358            }
359
360            assert_eq!(
361                *logger.lock().unwrap(),
362                vec![
363                    "FooDataSrc::new 1",
364                    "BarDataSrc::new 2",
365                    "FooDataSrc::setup_async 1",
366                    "BarDataSrc::setup_async 2",
367                    "FooDataSrc::create_data_src_async 1",
368                    "FooDataConn::new 1",
369                    "FooDataConn::get_text_async 1",
370                    "BarDataSrc::create_data_src_async 2",
371                    "BarDataConn::new 2",
372                    "BarDataConn::set_text_async 2",
373                    "FooDataConn::get_text_async 1",
374                    "BarDataConn::set_text_async 2",
375                    "BarDataConn.text = hello",
376                    "BarDataConn::close 2",
377                    "BarDataConn::drop 2",
378                    "FooDataConn::close 1",
379                    "FooDataConn::drop 1",
380                    "BarDataSrc.text = ", // because not committed
381                    "BarDataSrc::close 2",
382                    "BarDataSrc::drop 2",
383                    "FooDataSrc::close 1",
384                    "FooDataSrc::drop 1",
385                ],
386            );
387        }
388    }
389
390    mod test_txn_async_method {
391        use super::*;
392        use crate::tokio::logic;
393        use override_macro::{overridable, override_with};
394
395        #[overridable(mod = test_txn_async_method)]
396        trait SampleAsyncData {
397            async fn get_value_async(&mut self) -> errs::Result<String>;
398            async fn set_value_async(&mut self, v: &str) -> errs::Result<()>;
399        }
400
401        async fn sample_logic_async(data: &mut (impl SampleAsyncData + Send)) -> errs::Result<()> {
402            let v = data.get_value_async().await?;
403            let _ = data.set_value_async(&v).await;
404            let v = data.get_value_async().await?;
405            let _ = data.set_value_async(&v).await;
406            Ok(())
407        }
408
409        #[overridable(mod = test_txn_async_method)]
410        trait FooDataAcc: DataAcc {
411            async fn get_value_async(&mut self) -> errs::Result<String> {
412                let conn = self.get_data_conn_async::<FooDataConn>("foo").await?;
413                Ok(conn.get_text_async().await)
414            }
415        }
416
417        impl FooDataAcc for DataHub {}
418
419        #[overridable(mod = test_txn_async_method)]
420        trait BarDataAcc: DataAcc {
421            async fn set_value_async(&mut self, text: &str) -> errs::Result<()> {
422                let conn = self.get_data_conn_async::<BarDataConn>("bar").await?;
423                conn.set_text_async(text).await;
424                Ok(())
425            }
426        }
427
428        impl BarDataAcc for DataHub {}
429
430        #[override_with(test_txn_async_method::FooDataAcc, test_txn_async_method::BarDataAcc)]
431        impl test_txn_async_method::SampleAsyncData for DataHub {}
432
433        #[tokio::test]
434        async fn test_logic() {
435            let logger = Arc::new(Mutex::new(Vec::new()));
436
437            {
438                let mut data = DataHub::new();
439
440                data.uses("foo", FooDataSrc::new(1, "hello", logger.clone(), false));
441                data.uses("bar", BarDataSrc::new(2, logger.clone()));
442
443                if let Err(_) = data.txn_async(logic!(sample_logic_async)).await {
444                    panic!();
445                }
446            }
447
448            assert_eq!(
449                *logger.lock().unwrap(),
450                vec![
451                    "FooDataSrc::new 1",
452                    "BarDataSrc::new 2",
453                    "FooDataSrc::setup_async 1",
454                    "BarDataSrc::setup_async 2",
455                    "FooDataSrc::create_data_src_async 1",
456                    "FooDataConn::new 1",
457                    "FooDataConn::get_text_async 1",
458                    "BarDataSrc::create_data_src_async 2",
459                    "BarDataConn::new 2",
460                    "BarDataConn::set_text_async 2",
461                    "FooDataConn::get_text_async 1",
462                    "BarDataConn::set_text_async 2",
463                    "FooDataConn::pre_commit_async 1",
464                    "BarDataConn::pre_commit_async 2",
465                    "FooDataConn::commit_async 1",
466                    "BarDataConn::commit_async 2",
467                    "FooDataConn::post_commit_async 1",
468                    "BarDataConn::post_commit_async 2",
469                    "BarDataConn.text = hello",
470                    "BarDataConn::close 2",
471                    "BarDataConn::drop 2",
472                    "FooDataConn::close 1",
473                    "FooDataConn::drop 1",
474                    "BarDataSrc.text = hello", // because committed
475                    "BarDataSrc::close 2",
476                    "BarDataSrc::drop 2",
477                    "FooDataSrc::close 1",
478                    "FooDataSrc::drop 1",
479                ],
480            );
481        }
482    }
483}
sabi/tokio/data_acc.rs

sabi/tokio/
data_acc.rs
