detector 0.4.2

A set of types for service registration and discovery.
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

use crate::{Meta, MetaKey, Service, ServiceKey, ServiceStatus};

/// 监听事件
#[derive(Clone, Debug)]
pub enum ServiceEvent {
    /// 变化
    Changed(Service),
    /// 删除
    Deleted(ServiceKey),
}

#[derive(Clone, Debug)]
pub enum MetaEvent {
    /// 变化
    Changed(Meta),
    /// 删除  
    Deleted(MetaKey),
}

/// 事件接收者
pub struct WatchRx<T> {
    receiver: tokio::sync::mpsc::UnboundedReceiver<T>,
}

impl<T> WatchRx<T> {
    /// 如果监听被关闭则会返回none
    pub async fn recv(&mut self) -> Option<T> {
        self.receiver.recv().await
    }
}

/// 事件发送者
pub struct WatchTx<T> {
    sender: tokio::sync::mpsc::UnboundedSender<T>,
}

impl<T> WatchTx<T> {
    /// 通道被废弃会send失败
    pub fn send(&self, event: T) -> Result<(), T> {
        self.sender.send(event).map_err(|e| e.0)
    }
}

pub fn build_watch<T>() -> (WatchTx<T>, WatchRx<T>) {
    let (sender, receiver) = tokio::sync::mpsc::unbounded_channel();
    (WatchTx { sender }, WatchRx { receiver })
}

/// 自己状态接受者
#[derive(Clone, Debug)]
pub struct OneselfRx {
    receiver: tokio::sync::watch::Receiver<ServiceStatus>,
}

impl OneselfRx {
    /// 返回失败则说明通道被关闭了
    pub async fn changed(&mut self) -> Result<ServiceStatus, ()> {
        let _ = self.receiver.changed().await.map_err(|_| ())?;
        let status = *self.receiver.borrow();
        Ok(status)
    }

    /// 查看最新的状态值
    pub fn status(&self) -> ServiceStatus {
        *self.receiver.borrow()
    }
}

#[derive(Clone)]
pub struct OneselfTx {
    sender: tokio::sync::watch::Sender<ServiceStatus>,
    receiver: tokio::sync::watch::Receiver<ServiceStatus>,
}

impl OneselfTx {
    /// 通道被废弃会send失败
    pub fn send(&self, status: ServiceStatus) -> Result<(), ServiceStatus> {
        self.sender.send(status).map_err(|e| e.0)
    }

    pub fn oneself_rx(&self) -> OneselfRx {
        OneselfRx {
            receiver: self.receiver.clone(),
        }
    }

    pub fn count(&self) -> usize {
        self.sender.sender_count()
    }
}

pub fn build_oneself(status: ServiceStatus) -> (OneselfTx, OneselfRx) {
    let (sender, receiver) = tokio::sync::watch::channel(status);
    (
        OneselfTx {
            sender,
            receiver: receiver.clone(),
        },
        OneselfRx { receiver },
    )
}