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
126
127
#[cfg(test)]
mod tests;
use crate::worker::{Worker, WorkerSettings};
use async_executor::Executor;
use async_oneshot::Sender;
use event_listener_primitives::{Bag, HandlerId};
use futures_lite::future;
use log::debug;
use std::sync::Arc;
use std::{fmt, io};
#[derive(Default)]
struct Handlers {
new_worker: Bag<Box<dyn Fn(&Worker) + Send + Sync>>,
}
struct Inner {
executor: Arc<Executor<'static>>,
handlers: Handlers,
_stop_sender: Option<Sender<()>>,
}
#[derive(Clone)]
pub struct WorkerManager {
inner: Arc<Inner>,
}
impl fmt::Debug for WorkerManager {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("WorkerManager").finish()
}
}
impl Default for WorkerManager {
fn default() -> Self {
Self::new()
}
}
impl WorkerManager {
pub fn new() -> Self {
let executor = Arc::new(Executor::new());
let (stop_sender, stop_receiver) = async_oneshot::oneshot::<()>();
{
let executor = Arc::clone(&executor);
std::thread::spawn(move || {
let _ = future::block_on(executor.run(stop_receiver));
});
}
let handlers = Handlers::default();
let inner = Arc::new(Inner {
executor,
handlers,
_stop_sender: Some(stop_sender),
});
Self { inner }
}
pub fn with_executor(executor: Arc<Executor<'static>>) -> Self {
let handlers = Handlers::default();
let inner = Arc::new(Inner {
executor,
handlers,
_stop_sender: None,
});
Self { inner }
}
pub async fn create_worker(&self, worker_settings: WorkerSettings) -> io::Result<Worker> {
debug!("create_worker()");
let worker = Worker::new(
Arc::clone(&self.inner.executor),
worker_settings,
self.clone(),
)
.await?;
self.inner.handlers.new_worker.call(|callback| {
callback(&worker);
});
Ok(worker)
}
pub fn on_new_worker<F: Fn(&Worker) + Send + Sync + 'static>(&self, callback: F) -> HandlerId {
self.inner.handlers.new_worker.add(Box::new(callback))
}
}