tasking 0.4.0

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
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125

use std::{
  collections::{HashMap, HashSet},
  future::Future,
  hash::Hash,
  sync::{Arc, Mutex},
};

use crate::{AbortReceiver, Task, TaskDescriptor, TaskOptions, task_counter::TaskCounter};

pub struct TaskHub<TTaskKey, TTask, TTaskDescriptor> {
  name_prefix: String,
  task: Arc<Mutex<TTask>>,
  task_options: TaskOptions,
  task_map: Arc<Mutex<HashMap<TTaskKey, Arc<Task<TTask, TTaskDescriptor>>>>>,
  task_counter: Arc<TaskCounter>,
}

impl<
  TTaskKey: Clone + Eq + Hash + ToString + Send + 'static,
  TTask: FnMut(TTaskDescriptor, AbortReceiver) -> TTaskFuture + Send + 'static,
  TTaskReturn: Into<anyhow::Result<()>>,
  TTaskFuture: Future<Output = TTaskReturn> + Send + 'static,
  TTaskDescriptor: TaskDescriptor,
> TaskHub<TTaskKey, TTask, TTaskDescriptor>
{
  pub fn new(name_prefix: impl Into<String>, task: TTask, task_options: TaskOptions) -> Self {
    Self::new_with_counter(
      name_prefix,
      task,
      task_options,
      Arc::new(TaskCounter::new()),
    )
  }

  pub fn new_with_counter(
    name_prefix: impl Into<String>,
    task: TTask,
    task_options: TaskOptions,
    task_counter: Arc<TaskCounter>,
  ) -> Self {
    Self {
      name_prefix: name_prefix.into(),
      task: Arc::new(Mutex::new(task)),
      task_options,
      task_map: Arc::new(Mutex::new(HashMap::new())),
      task_counter,
    }
  }

  pub fn prune_inactive(&self) {
    self
      .task_map
      .lock()
      .unwrap()
      .retain(|_, task| task.is_active());
  }

  pub async fn update(&self, descriptors: impl IntoIterator<Item = (TTaskKey, TTaskDescriptor)>) {
    self.update_tasks(descriptors, false).await;
  }

  pub async fn merge(&self, descriptors: impl IntoIterator<Item = (TTaskKey, TTaskDescriptor)>) {
    self.update_tasks(descriptors, true).await;
  }

  pub async fn clear(&self) {
    self.task_map.lock().unwrap().clear();

    self.task_counter.cleared().await;
  }

  async fn update_tasks(
    &self,
    descriptors: impl IntoIterator<Item = (TTaskKey, TTaskDescriptor)>,
    keep_existing_tasks: bool,
  ) {
    let descriptors = descriptors.into_iter().collect::<Vec<_>>();

    let key_set = descriptors
      .iter()
      .map(|(key, _)| key)
      .cloned()
      .collect::<HashSet<_>>();

    if !keep_existing_tasks {
      self
        .task_map
        .lock()
        .unwrap()
        .retain(|key, _| key_set.contains(key));
    }

    let futures = descriptors
      .into_iter()
      .map(|(key, descriptor)| self.add_task(key, descriptor))
      .collect::<Vec<_>>();

    futures::future::join_all(futures).await;
  }

  async fn add_task(&self, key: TTaskKey, descriptor: TTaskDescriptor) {
    let task = self
      .task_map
      .lock()
      .unwrap()
      .entry(key.clone())
      .or_insert_with(|| {
        Arc::new(
          Task::new_internal(
            format!(
              "{prefix}:{key}",
              prefix = self.name_prefix,
              key = key.to_string()
            ),
            self.task.clone(),
            self.task_options.clone(),
          )
          .with_counter(self.task_counter.clone()),
        )
      })
      .clone();

    task.update(descriptor).await;
  }
}