axum-cometd 0.9.3

Framework for CometD server creation
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

mod client_timeout;

use crate::{
    messages::SubscriptionMessage,
    types::{ClientId, ClientReceiver, CookieId},
    LongPollingServiceContext,
};
use core::{fmt::Debug, time::Duration};
use std::sync::Arc;
use tokio::sync::{
    mpsc::{error::SendError, Receiver, Sender},
    Mutex, Notify,
};

#[derive(Debug)]
pub(crate) struct ClientSender {
    cookie_id: CookieId,
    signals: Arc<Signals>,
    tx: Sender<SubscriptionMessage>,
    rx: Arc<Mutex<Receiver<SubscriptionMessage>>>,
}

#[derive(Debug, Default)]
pub(crate) struct Signals {
    pub(crate) stop_signal: Notify,
    pub(crate) start_timeout: Notify,
    pub(crate) cancel_timeout: Notify,
}

impl ClientSender {
    #[inline]
    pub(crate) fn create<AdditionalData, CustomData>(
        context: Arc<LongPollingServiceContext<AdditionalData, CustomData>>,
        cookie_id: CookieId,
        client_id: ClientId,
        timeout: Duration,
        tx: Sender<SubscriptionMessage>,
        rx: Receiver<SubscriptionMessage>,
    ) -> Self
    where
        AdditionalData: Send + Sync + 'static,
        CustomData: Send + Sync + 'static,
    {
        let signals = Arc::new(Signals::default());
        let rx = Arc::new(Mutex::new(rx));

        client_timeout::spawn(context, client_id, timeout, Arc::clone(&signals));

        signals.start_timeout.notify_waiters();

        Self {
            cookie_id,
            signals,
            tx,
            rx,
        }
    }

    #[inline(always)]
    pub(crate) const fn cookie_id(&self) -> CookieId {
        self.cookie_id
    }

    #[inline]
    pub(crate) fn subscribe(&self) -> ClientReceiver {
        self.signals.cancel_timeout.notify_waiters();
        ClientReceiver::new(Arc::clone(&self.signals), Arc::clone(&self.rx))
    }

    #[inline(always)]
    pub(crate) async fn send(
        &self,
        msg: SubscriptionMessage,
    ) -> Result<(), SendError<SubscriptionMessage>> {
        self.tx.send(msg).await
    }
}

impl Drop for ClientSender {
    fn drop(&mut self) {
        self.signals.stop_signal.notify_one();
    }
}