axum-cometd 0.5.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
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55

use crate::{
    context::{Channel, LongPollingServiceContext},
    messages::SubscriptionMessage,
};
use serde_json::Value as JsonValue;
use std::sync::Arc;
use tokio::sync::mpsc;

pub(crate) fn spawn(
    channel: String,
    mut rx: mpsc::Receiver<JsonValue>,
    inner: Arc<LongPollingServiceContext>,
) {
    tokio::task::spawn(async move {
        while let Some(msg) = rx.recv().await {
            tracing::debug!(
                channel = channel,
                "`{channel}` channel got message: `{msg:?}`."
            );

            let client_id_channels = inner.client_id_senders.read().await;

            for (client_id, client_channel) in inner
                .channels_data
                .read()
                .await
                .get(&channel)
                .into_iter()
                .flat_map(Channel::client_ids)
                .filter_map(|client_id| client_id_channels.get(client_id).map(|v| (client_id, v)))
            {
                tracing::trace!(
                    client_id = %client_id,
                    channel = channel,
                    "Message `{msg:?}` from channel `{channel}` was sent to client `{client_id}`."
                );

                if client_channel
                    .send(SubscriptionMessage {
                        channel: channel.clone(),
                        msg: msg.clone(),
                    })
                    .await
                    .is_err()
                {
                    tracing::error!(
                        client_id = %client_id,
                        channel = channel,
                        "Channel was closed!"
                    );
                }
            }
        }
    });
}