Method subscribe

nodedb/src/control/pubsub.rs:243–274

Subscribe to a topic. Returns (subscription_id, receiver, backlog). `since_seq`: replay messages starting from this sequence number. 0 = no replay, just new messages.

(
        &self,
        topic_name: &str,
        since_seq: u64,
    )

    /// `since_seq`: replay messages starting from this sequence number.
    /// 0 = no replay, just new messages.
    pub fn subscribe(
        &self,
        topic_name: &str,
        since_seq: u64,
    ) -> Result<
        (
            u64,
            tokio::sync::mpsc::Receiver<PubSubMessage>,
            Vec<PubSubMessage>,
        ),
        TopicError,
    > {
        let topics = super::lock_utils::read_or_recover(self.topics.read(), "topics");
        let topic_mutex = topics
            .get(topic_name)
            .ok_or_else(|| TopicError::NotFound(topic_name.to_string()))?;
        let mut topic = super::lock_utils::lock_or_recover(topic_mutex.lock(), "topic");

        let sub_id = self.next_sub_id.fetch_add(1, Ordering::Relaxed);
        let (tx, rx) = tokio::sync::mpsc::channel(256);

        let backlog = topic.get_backlog(since_seq);

        topic.subscribers.insert(sub_id, tx);
        debug!(
            topic = topic_name,
            sub_id,
            backlog = backlog.len(),
            "subscribed"
        );
        Ok((sub_id, rx, backlog))
    }

    /// Subscribe to a topic as a consumer group member.
    ///