dhatu 0.2.2

dhatu core libraries
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65

use std::{collections::VecDeque, sync::Arc};

use tokio::sync::Mutex;

type InnerPool<T> = VecDeque<T>;

pub struct ThreadSafeQueue<T> {
    inner: Arc<Mutex<InnerPool<T>>>,
}

impl<T> Clone for ThreadSafeQueue<T> {
    fn clone(&self) -> Self {
        Self {
            inner: self.inner.clone(),
        }
    }
}

impl<T> ThreadSafeQueue<T> {
    pub fn new() -> Self {
        let inner = VecDeque::<T>::new();
        let inner = Arc::new(Mutex::new(inner));

        Self { inner }
    }

    pub async fn get(&mut self) -> Option<T> {
        let mut inner = self.inner.lock().await;
        inner.pop_front()
    }

    pub async fn add(&mut self, data: T) {
        let mut inner = self.inner.lock().await;
        inner.push_back(data)
    }
}

// #[cfg(test)]
// mod tests {
//     use super::*;
//     #[actix::test]
//     async fn should_queue_correctly() {
//         let mut queue = ThreadSafeQueue::new();

//         let expected_first_batch = vec![1];
//         let expected_second_batch = vec![2];
//         let expected_third_batch = vec![3];

//         queue.add(expected_first_batch.clone()).await;
//         queue.add(expected_second_batch.clone()).await;
//         queue.add(expected_third_batch.clone()).await;

//         let actual_first_batch = queue.get().await.unwrap();
//         let actual_second_batch = queue.get().await.unwrap();
//         let actual_third_batch = queue.get().await.unwrap();

//         assert_eq!(expected_first_batch[0], actual_first_batch[0]);
//         assert_eq!(expected_second_batch[0], actual_second_batch[0]);
//         assert_eq!(expected_third_batch[0], actual_third_batch[0]);

//         let batch = queue.get().await;

//         assert!(batch.is_none());
//     }
// }