Function handle_psubscribe

nodedb/src/control/server/resp/handler_pubsub.rs:128–227

Handle PSUBSCRIBE command: pattern-based subscription. Like SUBSCRIBE, but channel names are matched using glob patterns (`*` matches any string, `?` matches one character). Push messages use the `pmessage` type with 4 elements: `[pmessage, pattern, channel, payload]`.

(
    cmd: &RespCommand,
    session: &RespSession,
    state: &SharedState,
    stream: &mut ConnStream,
)

/// (`*` matches any string, `?` matches one character). Push messages
/// use the `pmessage` type with 4 elements: `[pmessage, pattern, channel, payload]`.
pub async fn handle_psubscribe(
    cmd: &RespCommand,
    session: &RespSession,
    state: &SharedState,
    stream: &mut ConnStream,
) -> crate::Result<()> {
    if cmd.argc() < 1 {
        let resp = RespValue::err("ERR wrong number of arguments for 'psubscribe' command");
        let bytes = resp.to_bytes();
        stream
            .write_all(&bytes)
            .await
            .map_err(|e| crate::Error::Bridge {
                detail: format!("RESP write: {e}"),
            })?;
        return Ok(());
    }

    let patterns: Vec<String> = cmd
        .args
        .iter()
        .filter_map(|a| std::str::from_utf8(a).ok().map(|s| s.to_string()))
        .collect();

    let mut subscription = state.change_stream.subscribe(None, Some(session.tenant_id));

    // Send psubscribe confirmation for each pattern.
    for (i, pattern) in patterns.iter().enumerate() {
        let confirm = RespValue::array(vec![
            RespValue::bulk_str("psubscribe"),
            RespValue::bulk_str(pattern),
            RespValue::integer((i + 1) as i64),
        ]);
        let bytes = confirm.to_bytes();
        stream
            .write_all(&bytes)
            .await
            .map_err(|e| crate::Error::Bridge {
                detail: format!("RESP write: {e}"),
            })?;
    }

    debug!(
        patterns = ?patterns,
        "RESP PSUBSCRIBE: entering pattern subscription mode"
    );

    // Subscription loop with glob matching.
    loop {
        match subscription.receiver.recv().await {
            Ok(event) => {
                // Check if event.collection matches ANY subscribed pattern.
                let matched_pattern = patterns.iter().find(|p| {
                    crate::engine::kv::scan::glob_match(p.as_bytes(), event.collection.as_bytes())
                });
                let Some(pattern) = matched_pattern else {
                    continue;
                };