Method subscribe_group

nodedb/src/control/pubsub.rs:281–320

Subscribe to a topic as a consumer group member. Messages are delivered to exactly ONE member of the group (round-robin). Multiple subscribers with the same `group_name` share the load. Returns (subscription_id, receiver, backlog).

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

    /// Multiple subscribers with the same `group_name` share the load.
    /// Returns (subscription_id, receiver, backlog).
    pub fn subscribe_group(
        &self,
        topic_name: &str,
        group_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);

        let group = topic
            .consumer_groups
            .entry(group_name.to_string())
            .or_insert_with(ConsumerGroup::new);
        group.members.push((sub_id, tx));

        debug!(
            topic = topic_name,
            group = group_name,
            sub_id,
            members = group.members.len(),
            backlog = backlog.len(),
            "subscribed to consumer group"
        );
        Ok((sub_id, rx, backlog))
    }

    /// Unsubscribe from a topic.
    pub fn unsubscribe(&self, topic_name: &str, sub_id: u64) {