calimero-node-primitives 0.11.0-rc.2

Core Calimero infrastructure and tools
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

use std::collections::HashSet;
use std::sync::Arc;

use calimero_network_primitives::client::NetworkClient;
use libp2p::gossipsub::IdentTopic;
use tokio::sync::RwLock;
use tracing::{debug, info};

/// Tracks gossipsub topic subscriptions with deduplication.
#[derive(Clone, Debug)]
pub struct TopicManager {
    network_client: NetworkClient,
    subscribed: Arc<RwLock<HashSet<String>>>,
}

impl TopicManager {
    pub fn new(network_client: NetworkClient) -> Self {
        Self {
            network_client,
            subscribed: Arc::new(RwLock::new(HashSet::new())),
        }
    }

    /// Subscribe to a topic if not already subscribed.
    pub async fn ensure_subscribed(&self, topic: &str) -> eyre::Result<()> {
        {
            let subs = self.subscribed.read().await;
            if subs.contains(topic) {
                debug!(topic, "already subscribed, skipping");
                return Ok(());
            }
        }
        let ident_topic = IdentTopic::new(topic);
        let _ignored = self.network_client.subscribe(ident_topic).await?;
        self.subscribed.write().await.insert(topic.to_owned());
        info!(topic, "subscribed to topic");
        Ok(())
    }

    /// Unsubscribe from a topic.
    pub async fn unsubscribe(&self, topic: &str) -> eyre::Result<()> {
        let ident_topic = IdentTopic::new(topic);
        let _ignored = self.network_client.unsubscribe(ident_topic).await?;
        self.subscribed.write().await.remove(topic);
        info!(topic, "unsubscribed from topic");
        Ok(())
    }

    /// Check if subscribed to a topic.
    pub async fn is_subscribed(&self, topic: &str) -> bool {
        self.subscribed.read().await.contains(topic)
    }

    /// List all subscribed topics.
    pub async fn subscribed_topics(&self) -> Vec<String> {
        self.subscribed.read().await.iter().cloned().collect()
    }
}