rspack_storage 0.7.11

rspack cache storage
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
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91

use std::{fmt::Debug, future::Future, sync::LazyLock};

use futures::{FutureExt, future::BoxFuture};
use tokio::sync::mpsc::{UnboundedSender, unbounded_channel};

pub struct TaskQueue(LazyLock<UnboundedSender<BoxFuture<'static, ()>>>);

impl Debug for TaskQueue {
  fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
    write!(f, "TaskQueue {{... }}")
  }
}

impl Default for TaskQueue {
  fn default() -> Self {
    Self::new()
  }
}

impl TaskQueue {
  pub fn new() -> Self {
    TaskQueue(LazyLock::new(|| {
      let (tx, mut rx) = unbounded_channel();
      tokio::spawn(async move {
        while let Some(future) = rx.recv().await {
          future.await
        }
      });

      tx
    }))
  }

  pub fn add_task(&self, task: impl Future<Output = ()> + Send + 'static) {
    self.0.send(task.boxed()).expect("should add task");
  }
}

#[cfg(test)]
mod tests {
  use std::{sync::Arc, time::Duration};

  use tokio::sync::{Mutex, oneshot};

  use crate::pack::manager::queue::TaskQueue;

  async fn test_task_queue() {
    let queue = TaskQueue::new();
    let (tx_0, rx_0) = oneshot::channel();
    let (tx_1, rx_1) = oneshot::channel();
    let (tx_2, rx_2) = oneshot::channel();

    let inc = Arc::new(Mutex::new(0_usize));

    let inc_0 = inc.clone();
    queue.add_task(Box::pin(async move {
      tokio::time::sleep(Duration::from_millis(30)).await;
      let mut inc = inc_0.lock().await;
      *inc += 1;
      tx_0.send(*inc).unwrap();
    }));
    let inc_1 = inc.clone();
    queue.add_task(Box::pin(async move {
      tokio::time::sleep(Duration::from_millis(20)).await;
      let mut inc = inc_1.lock().await;
      *inc += 1;
      tx_1.send(*inc).unwrap();
    }));
    let inc_2 = inc.clone();
    queue.add_task(Box::pin(async move {
      tokio::time::sleep(Duration::from_millis(10)).await;
      let mut inc = inc_2.lock().await;
      *inc += 1;
      tx_2.send(*inc).unwrap();
    }));

    assert_eq!(rx_0.await, Ok(1));
    assert_eq!(rx_1.await, Ok(2));
    assert_eq!(rx_2.await, Ok(3));
  }

  #[test]
  #[cfg_attr(miri, ignore)]
  fn should_add_task_to_queue() {
    let rt = tokio::runtime::Runtime::new().unwrap();

    rt.block_on(async {
      test_task_queue().await;
    });
  }
}