Method publish

nodedb/src/control/pubsub.rs:180–237

Publish a message to a topic. Returns `(sequence_number, receivers)` where `receivers` is the count of broadcast subscribers + consumer groups that received the message.

(
        &self,
        topic_name: &str,
        payload: String,
        publisher: &str,
    )

    /// Returns `(sequence_number, receivers)` where `receivers` is the count of
    /// broadcast subscribers + consumer groups that received the message.
    pub fn publish(
        &self,
        topic_name: &str,
        payload: String,
        publisher: &str,
    ) -> Result<(u64, usize), 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 seq = topic.next_seq;
        topic.next_seq += 1;

        let msg = PubSubMessage {
            seq,
            payload,
            timestamp_ms: std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .map(|d| d.as_millis() as u64)
                .unwrap_or(0),
            publisher: publisher.to_string(),
        };

        if topic.messages.len() >= topic.max_messages {
            topic.messages.remove(0);
        }
        topic.messages.push(msg.clone());

        // Deliver to broadcast subscribers. Remove dead channels.
        let mut dead = Vec::new();
        for (&sub_id, sender) in &topic.subscribers {
            if sender.try_send(msg.clone()).is_err() {
                debug!(
                    topic = topic_name,
                    sub_id, "subscriber channel full or closed; removing"
                );
                dead.push(sub_id);
            }
        }
        for sub_id in dead {
            topic.subscribers.remove(&sub_id);
        }

        // Count live broadcast subscribers (after removing dead ones).
        let broadcast_count = topic.subscribers.len();

        // Deliver to consumer groups (one member per group, round-robin).
        let group_count = topic.consumer_groups.len();
        for group in topic.consumer_groups.values_mut() {
            group.deliver(&msg);
        }

        let receivers = broadcast_count + group_count;
        debug!(topic = topic_name, seq, receivers, "message published");
        Ok((seq, receivers))
    }