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

use super::*;

// Here for future-proofing, in case we want to switch over to another mpsc implementation
pub type Sender<T> = std::sync::mpsc::Sender<T>;
pub type SyncSender<T> = std::sync::mpsc::SyncSender<T>;
pub type Receiver<T> = std::sync::mpsc::Receiver<T>;
pub type TryRecvError = std::sync::mpsc::TryRecvError;
pub type RecvError = std::sync::mpsc::RecvError;
pub type SendError<T> = std::sync::mpsc::SendError<T>;
pub type TrySendError<T> = std::sync::mpsc::TrySendError<T>;

/// A message sent from the manager to the workers
#[derive(Clone, Debug, PartialEq)]
pub enum DownMsg<Down: Send> {
    Stop,
    Pause,
    Continue,
    Other(Down)
}

/// A message sent from a worker to the manager; contains the timestamp of its creation to
/// allow RecvBurstIterator to stop early
#[derive(Clone, Debug)]
pub struct UpMsg<Up: Send> {
    time: Instant,
    msg: Up
}

/// A wrapper around Sender<UpMsg<Up>>. This type implements `!Send`,
/// as RecvAllIterator depends on this type being dropped whenever the thread holding it stops
#[derive(Clone, Debug)]
pub struct WorkerSender<Up: Send> {
    sender: SyncSender<UpMsg<Up>>,
}

impl<Up: Send> UpMsg<Up> {
    pub fn new(msg: Up) -> Self {
        Self {
            time: Instant::now(),
            msg
        }
    }

    pub fn time(&self) -> Instant {
        self.time
    }

    pub fn get(self) -> Up {
        self.msg
    }
}

impl<Up: Send> WorkerSender<Up> {
    pub fn send(&self, msg: Up) -> Result<(), SendError<Up>> {
        self.sender.send(UpMsg::new(msg)).map_err(|e| std::sync::mpsc::SendError(e.0.get()))
    }

    pub fn try_send(&self, msg: Up) -> Result<(), TrySendError<Up>> {
        use std::sync::mpsc::TrySendError;

        self.sender.try_send(UpMsg::new(msg)).map_err(|e| {
            match e {
                TrySendError::Full(x) => TrySendError::Full(x.get()),
                TrySendError::Disconnected(x) => TrySendError::Disconnected(x.get())
            }
        })
    }
}

impl<Up: Send> !Send for WorkerSender<Up> {}

impl<Up: Send> From<SyncSender<UpMsg<Up>>> for WorkerSender<Up> {
    fn from(sender: SyncSender<UpMsg<Up>>) -> Self {
        Self {
            sender
        }
    }
}

/// A wrapper around Receiver<DownMsg<Down>>
pub type WorkerReceiver<Down> = Receiver<DownMsg<Down>>;