axum-cometd 0.1.0

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

use async_trait::async_trait;
use std::time::Duration;
use tokio::{sync::broadcast, time};

#[async_trait]
pub trait ReceiverExt<T> {
    async fn recv_ignore_lagged(&mut self) -> Option<T>;
    async fn recv_ignore_lagged_timeout(
        &mut self,
        duration: Duration,
    ) -> Result<Option<T>, time::error::Elapsed>;
}

#[async_trait]
impl<T> ReceiverExt<T> for broadcast::Receiver<T>
where
    T: Clone + Send,
{
    #[inline]
    async fn recv_ignore_lagged(&mut self) -> Option<T> {
        loop {
            match self.recv().await {
                Ok(data) => break Some(data),
                Err(broadcast::error::RecvError::Lagged(num)) => {
                    tracing::warn!("RecvError::Laged({num})");
                    continue;
                }
                Err(broadcast::error::RecvError::Closed) => break None,
            }
        }
    }

    #[inline]
    async fn recv_ignore_lagged_timeout(
        &mut self,
        duration: Duration,
    ) -> Result<Option<T>, time::error::Elapsed> {
        time::timeout(duration, self.recv_ignore_lagged()).await
    }
}