Redis RS PubSub Channel Subscribe

use redis::{Client, RedisResult};
use std::time::Duration;

async fn listen(
    url: &str,
    channels: &[&str],
    max_messages: usize,
) -> RedisResult<Vec<String>> {
    let client = Client::open(url)?;
    let mut received = Vec::new();

    let mut pubsub = client.get_async_pubsub().await?;
    for channel in channels {
        pubsub.subscribe(channel).await?;
    }

    // 收满或超时即退出：演示可中断的订阅循环
    let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
    while received.len() < max_messages {
        let next = tokio::time::timeout_at(deadline, pubsub.on_message()).await;
        match next {
            Ok(Ok(msg)) => {
                let channel = msg.get_channel_name();
                let payload: String = msg.get_payload()?;
                received.push(format!("{channel}: {payload}"));
            }
            _ => break, // 超时或错误：跳出循环
        }
    }
    Ok(received)
}

#[tokio::main]
async fn main() -> RedisResult<()> {
    let url = "redis://127.0.0.1:6379";

    // 发布端：独立连接直接 PUBLISH
    let publisher = Client::open(url)?;
    let mut pub_con = publisher.get_multiplexed_connection().await?;

    let listener = tokio::spawn(listen(url, &["events:billing", "events:auth"], 3));

    tokio::time::sleep(Duration::from_millis(100)).await;
    let _: () = redis::cmd("PUBLISH")
        .arg("events:billing")
        .arg("invoice-settled")
        .query_async(&mut pub_con)
        .await?;
    let _: () = redis::cmd("PUBLISH")
        .arg("events:auth")
        .arg("user-login")
        .query_async(&mut pub_con)
        .await?;

    let messages = listener.await.unwrap()?;
    for msg in &messages {
        println!("{msg}");
    }
    Ok(())
}