tasking 0.3.3

Simple tasking utility that automatically replaces tasks based on descriptors.
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
92
93
94
95
96

use std::sync::{
  Arc,
  atomic::{AtomicUsize, Ordering},
};

use lits::duration;

use tasking::{TaskHub, TaskOptions};

#[tokio::test]
async fn taskhub_adds_and_removes_tasks_on_update() {
  let starts = Arc::new(AtomicUsize::new(0));
  let starts_c = starts.clone();

  let options = TaskOptions {
    restart_on_error: false,
    restart_interval: duration!("10ms"),
    abort_timeout: Some(duration!("100ms")),
  };

  let hub = TaskHub::new(
    "hub",
    move |_d: (), _abort| {
      let starts = starts_c.clone();

      async move {
        starts.fetch_add(1, Ordering::SeqCst);

        // run until aborted
        std::future::pending::<()>().await;

        Ok(())
      }
    },
    options,
  );

  // add two tasks
  hub
    .update(vec![("a".to_string(), ()), ("b".to_string(), ())])
    .await;
  tokio::time::sleep(duration!("20ms")).await;

  assert_eq!(starts.load(Ordering::SeqCst), 2);

  // update to only include "a" -> "b" should be removed
  hub.update(vec![("a".to_string(), ())]).await;
  tokio::time::sleep(duration!("20ms")).await;

  // still one task started for "a", total starts remains 2
  assert_eq!(starts.load(Ordering::SeqCst), 2);
}

#[tokio::test]
async fn taskhub_merge_keeps_running_tasks() {
  let starts = Arc::new(AtomicUsize::new(0));
  let starts_c = starts.clone();

  let options = TaskOptions {
    restart_on_error: false,
    restart_interval: duration!("10ms"),
    abort_timeout: Some(duration!("100ms")),
  };

  let hub = TaskHub::new(
    "hub-merge",
    move |_d: (), _abort| {
      let starts = starts_c.clone();

      async move {
        starts.fetch_add(1, Ordering::SeqCst);

        // run until aborted
        std::future::pending::<()>().await;

        Ok(())
      }
    },
    options,
  );

  // add two tasks
  hub
    .update(vec![("x".to_string(), ()), ("y".to_string(), ())])
    .await;
  tokio::time::sleep(duration!("20ms")).await;

  assert_eq!(starts.load(Ordering::SeqCst), 2);

  // merge with only x -> y should be kept running because merge preserves running tasks
  hub.merge(vec![("x".to_string(), ())]).await;
  tokio::time::sleep(duration!("20ms")).await;

  // no new starts should be created
  assert_eq!(starts.load(Ordering::SeqCst), 2);
}