cometd-client 0.3.6

Client for LongPolling CometD connections
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

use crate::{client::CometdClientInner, retry_with_advice, types::*};
use core::future::{ready, Future};
use serde::de::DeserializeOwned;
use std::sync::Arc;
use tokio::select;

#[inline(always)]
pub(crate) fn spawn(
    inner: CometdClientInner,
    cmd_rx: CmdReceiver,
    event_tx: EventSender<impl DeserializeOwned + Send + Sync + 'static>,
) {
    tokio::task::spawn(async move {
        let broadcast_event = |event| async {
            let _ = event_tx.broadcast(event).await;
        };

        if let Err(error) = retry_with_advice(
            inner.number_of_retries,
            || ready(Ok(())),
            || inner.handshake(),
        )
        .await
        {
            broadcast_event(CometdClientEvent::error(error)).await;
        } else {
            cmd_connect_loop(inner, cmd_rx, broadcast_event).await;
        }
    });
}

#[inline(always)]
async fn cmd_connect_loop<Msg, Fut>(
    inner: CometdClientInner,
    mut cmd_rx: CmdReceiver,
    broadcast_event: impl Fn(CometdClientEvent<Msg>) -> Fut,
) where
    Msg: DeserializeOwned,
    Fut: Future<Output = ()>,
{
    enum Res<Msg> {
        Left(Option<Command>),
        Right(CometdResult<Arc<[Data<Msg>]>>),
    }
    use Res::*;

    loop {
        let res = select! {
            biased;
            cmd = cmd_rx.recv() => Left(cmd),
            data = retry_with_advice(
                inner.number_of_retries,
                || inner.handshake(),
                || inner.connect::<Msg>(),
            ) => Right(data),
        };

        match res {
            Left(Some(Command::Subscribe(subscriptions))) => {
                if let Err(error) = retry_with_advice(
                    inner.number_of_retries,
                    || inner.handshake(),
                    || inner.subscribe(&subscriptions),
                )
                .await
                {
                    broadcast_event(CometdClientEvent::error(error)).await;
                    break;
                }
            }
            Right(Ok(data)) => broadcast_event(CometdClientEvent::Message(data)).await,
            // communication errors
            Left(None) => break,
            Right(Err(error)) => {
                broadcast_event(CometdClientEvent::error(error)).await;
                break;
            }
        }
    }

    if let Err(error) = inner.disconnect().await {
        broadcast_event(CometdClientEvent::error(error)).await;
    }
}