endpoint-libs 1.5.0

Common dependencies to be used with Pathscale projects, projects that use [endpoint-gen](https://github.com/pathscale/endpoint-gen), and projects that use honey_id-types.
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

#![allow(dead_code)]

use dashmap::DashMap;
use serde::Serialize;
use std::collections::HashMap;
use std::hash::Hash;

use crate::libs::toolbox::{ArcToolbox, RequestContext};

use super::{ConnectionId, WsResponseGeneric, WsStreamResponseGeneric};

pub struct SubscriberContext {
    pub ctx: RequestContext,
    pub stream_seq: u32,
}

#[derive(Default)]
pub struct Subscribers {
    pub subscribers: HashMap<ConnectionId, SubscriberContext>,
}

pub struct SubscribeManager<Key: Eq + Hash> {
    pub topics: DashMap<Key, Subscribers>,
}

impl<Key: Hash + Eq + Into<u32>> Default for SubscribeManager<Key> {
    fn default() -> Self {
        Self::new()
    }
}

impl<Key: Hash + Eq + Into<u32>> SubscribeManager<Key> {
    pub fn new() -> Self {
        Self {
            topics: DashMap::new(),
        }
    }

    pub fn add_topics(&self, topics: Vec<Key>) {
        for topic in topics {
            self.add_topic(topic);
        }
    }
    pub fn add_topic(&self, topic: Key) {
        self.topics.entry(topic).or_default();
    }
    pub fn subscribe_multi(&self, topics: Vec<Key>, ctx: RequestContext) {
        for topic in topics {
            self.subscribe(topic, ctx.clone());
        }
    }
    pub fn subscribe(&self, topic: Key, ctx: RequestContext) {
        let mut subscribers = self.topics.entry(topic).or_default();
        subscribers
            .subscribers
            .insert(ctx.connection_id, SubscriberContext { ctx, stream_seq: 0 });
    }
    pub fn unsubscribe_multi(&self, topics: Vec<Key>, connection_id: ConnectionId) {
        for topic in topics {
            self.unsubscribe(topic, connection_id);
        }
    }
    pub fn unsubscribe(&self, topic: Key, connection_id: ConnectionId) {
        if let Some(mut subscribers) = self.topics.get_mut(&topic) {
            subscribers.subscribers.remove(&connection_id);
        }
    }
    pub fn publish_with_filter(
        &self,
        toolbox: &ArcToolbox,
        topic: Key,
        msg: &impl Serialize,
        filter: impl Fn(&RequestContext) -> bool,
    ) {
        if let Some(mut topic_2) = self.topics.get_mut(&topic) {
            let data = serde_json::to_value(msg).unwrap();
            let mut dead_connections = vec![];
            let stream_code = topic.into();
            for sub in topic_2.subscribers.values_mut() {
                if !filter(&sub.ctx) {
                    continue;
                }
                let msg = WsResponseGeneric::Stream(WsStreamResponseGeneric {
                    original_seq: sub.ctx.seq,
                    method: sub.ctx.method,
                    stream_seq: sub.stream_seq,
                    stream_code,
                    data: data.clone(),
                });
                sub.stream_seq += 1;
                if !toolbox.send(sub.ctx.connection_id, msg) {
                    dead_connections.push(sub.ctx.connection_id);
                }
            }
            for conn_id in dead_connections {
                topic_2.subscribers.remove(&conn_id);
            }
        }
    }
    pub fn publish_to_all(&self, toolbox: &ArcToolbox, topic: Key, msg: &impl Serialize) {
        self.publish_with_filter(toolbox, topic, msg, |_| true)
    }
}