agner_reg/
reg.rs

1use std::sync::Arc;
2
3use agner_actors::ActorID;
4use tokio::sync::watch;
5
6pub fn new() -> (RegTx, RegRx) {
7    let (tx, rx) = watch::channel(None);
8    let tx = RegTx(Arc::new(tx));
9    let rx = RegRx(rx);
10    (tx, rx)
11}
12
13#[derive(Debug, Clone)]
14pub struct RegRx(watch::Receiver<State>);
15
16#[derive(Debug, Clone)]
17pub struct RegTx(RegTxInner);
18
19#[derive(Debug)]
20pub struct RegGuard {
21    tx_inner: RegTxInner,
22    actor_id: ActorID,
23}
24
25impl RegTx {
26    pub fn register(&self, actor_id: ActorID) -> RegGuard {
27        RegGuard::new(Arc::clone(&self.0), actor_id)
28    }
29}
30
31impl RegRx {
32    pub fn resolve(&self) -> Option<ActorID> {
33        *self.0.borrow()
34    }
35
36    pub async fn wait(&self) -> Option<ActorID> {
37        let mut rx = self.0.clone();
38        loop {
39            if let Some(actor_id) = *rx.borrow() {
40                break Some(actor_id)
41            }
42
43            if rx.changed().await.is_err() {
44                break None
45            }
46        }
47    }
48}
49
50type State = Option<ActorID>;
51type RegTxInner = Arc<watch::Sender<State>>;
52
53impl RegGuard {
54    fn new(tx_inner: RegTxInner, actor_id: ActorID) -> Self {
55        tx_inner.send_modify(|v| *v = Some(actor_id));
56        Self { tx_inner, actor_id }
57    }
58}
59
60impl Drop for RegGuard {
61    fn drop(&mut self) {
62        self.tx_inner.send_modify(|v| {
63            if matches!(v, Some(this) if *this == self.actor_id) {
64                *v = None;
65            }
66        });
67    }
68}
agner_reg/reg.rs

agner_reg/
reg.rs
