netidx 0.31.5

Secure, fast, pub/sub messaging
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

use futures::channel::oneshot;
use parking_lot::Mutex;
use poolshark::global::{GPooled, Pool};
use std::sync::Arc;
use std::{clone::Clone, mem, ops::Drop, result};

#[derive(Debug)]
struct BatchChannelInner<T: Send + Sync + 'static> {
    send_closed: bool,
    recv_closed: bool,
    notify: Option<oneshot::Sender<()>>,
    queue: GPooled<Vec<T>>,
    pool: Pool<Vec<T>>,
}

#[derive(Debug)]
struct BatchSenderInner<T: Send + Sync + 'static>(Arc<Mutex<BatchChannelInner<T>>>);

impl<T: Send + Sync + 'static> Drop for BatchSenderInner<T> {
    fn drop(&mut self) {
        self.0.lock().send_closed = true;
    }
}

#[derive(Debug)]
pub(crate) struct BatchSender<T: Send + Sync + 'static>(Arc<BatchSenderInner<T>>);

impl<T: Send + Sync + 'static> Clone for BatchSender<T> {
    fn clone(&self) -> Self {
        BatchSender(Arc::clone(&self.0))
    }
}

impl<T: Send + Sync + 'static> BatchSender<T> {
    pub(crate) fn send(&self, m: T) -> bool {
        let mut inner = self.0 .0.lock();
        if inner.recv_closed {
            false
        } else {
            inner.queue.push(m);
            if let Some(sender) = inner.notify.take() {
                let _: result::Result<_, _> = sender.send(());
            }
            true
        }
    }
}

#[derive(Debug)]
pub(crate) struct BatchReceiver<T: Send + Sync + 'static>(
    Arc<Mutex<BatchChannelInner<T>>>,
);

impl<T: Send + Sync + 'static> Drop for BatchReceiver<T> {
    fn drop(&mut self) {
        self.close()
    }
}

impl<T: Send + Sync + 'static> BatchReceiver<T> {
    pub(crate) fn close(&self) {
        let mut inner = self.0.lock();
        inner.recv_closed = true;
        inner.queue.clear();
        inner.notify = None;
    }

    pub(crate) fn len(&self) -> usize {
        self.0.lock().queue.len()
    }

    pub(crate) async fn recv(&self) -> Option<GPooled<Vec<T>>> {
        loop {
            let receiver = {
                let mut inner = self.0.lock();
                if inner.queue.len() > 0 {
                    let v = inner.pool.take();
                    return Some(mem::replace(&mut inner.queue, v));
                } else if inner.send_closed {
                    return None;
                } else {
                    let (tx, rx) = oneshot::channel();
                    inner.notify = Some(tx);
                    rx
                }
            };
            let _: result::Result<_, _> = receiver.await;
        }
    }
}

pub(crate) fn channel<T: Send + Sync + 'static>() -> (BatchSender<T>, BatchReceiver<T>) {
    let pool = Pool::new(1, 100_000);
    let inner = Arc::new(Mutex::new(BatchChannelInner {
        send_closed: false,
        recv_closed: false,
        notify: None,
        queue: pool.take(),
        pool,
    }));
    let sender = BatchSender(Arc::new(BatchSenderInner(inner.clone())));
    let receiver = BatchReceiver(inner);
    (sender, receiver)
}