sabi_redis 0.8.0

// Copyright (C) 2026 Takayuki Sato. All Rights Reserved.
// This program is free software under MIT License.
// See the file LICENSE in this distribution for more details.

use crate::retry_async::RetryAsync;
use futures::{stream::StreamExt, Future};
use redis::aio::PubSub;
use redis::{Client, ConnectionAddr, ConnectionInfo, ControlFlow, Msg, ToRedisArgs};
use std::fmt::Debug;

/// Errors related to asynchronous Redis Pub/Sub subscriber for standalone configuration.
#[derive(Debug)]
pub enum RedisPubSubSubscriberErrorAsync {
    /// Indicates that the connection address has already been used and cannot be reused.
    AddressAlreadyConsumed,
    /// Failed to open a Redis client with the specified address string.
    FailToOpenClientOfAddr {
        /// The Redis connection address string.
        addr: String,
    },
    /// Failed to open a Redis client with the specified `ConnectionAddr`.
    FailToOpenClientOfConnAddr {
        /// The Redis `ConnectionAddr`.
        conn_addr: ConnectionAddr,
    },
    /// Failed to open a Redis client with the specified `ConnectionInfo`.
    FailToOpenClientOfConnInfo {
        /// The Redis `ConnectionInfo`.
        conn_info: ConnectionInfo,
    },
    /// Failed to get an asynchronous Pub/Sub connection from the client.
    FailToGetAsyncPubSub {
        /// The Redis `ConnectionInfo`.
        conn_info: ConnectionInfo,
    },
    /// Failed to subscribe to the specified channels.
    FailToSubscribeToChannels {
        /// The Redis `ConnectionInfo`.
        conn_info: ConnectionInfo,
    },
    /// Failed to subscribe to the specified patterns.
    FailToSubscribeToChannelsWithPatterns {
        /// The Redis `ConnectionInfo`.
        conn_info: ConnectionInfo,
    },
    /// Failed to receive a message from the subscriber.
    FailToGetMessage {
        /// The Redis `ConnectionInfo`.
        conn_info: ConnectionInfo,
    },
}

/// A struct for subscribing to Redis channels and receiving messages asynchronously for standalone configuration.
pub struct RedisPubSubSubscriberAsync<A>
where
    A: ToRedisArgs,
{
    addr: Option<RedisConfig>,
    channels: Vec<A>,
    patterns: Vec<A>,
    retry: RetryAsync,
}

enum RedisConfig {
    String(String),
    ConnAddr(ConnectionAddr),
    ConnInfo(ConnectionInfo),
}

impl<A> RedisPubSubSubscriberAsync<A>
where
    A: ToRedisArgs,
{
    /// Creates a new `RedisPubSubSubscriberAsync` with a connection address string.
    ///
    /// # Arguments
    ///
    /// * `addr` - A string slice that holds the Redis connection address.
    ///
    /// # Returns
    ///
    /// A new instance of `RedisPubSubSubscriberAsync`.
    pub fn new<S>(addr: S) -> Self
    where
        S: AsRef<str>,
    {
        Self {
            addr: Some(RedisConfig::String(addr.as_ref().to_string())),
            channels: Vec::new(),
            patterns: Vec::new(),
            retry: RetryAsync::new(),
        }
    }

    /// Creates a new `RedisPubSubSubscriberAsync` with a `ConnectionAddr`.
    ///
    /// # Arguments
    ///
    /// * `conn_addr` - A Redis `ConnectionAddr`.
    ///
    /// # Returns
    ///
    /// A new instance of `RedisPubSubSubscriberAsync`.
    pub fn with_conn_addr(conn_addr: ConnectionAddr) -> Self {
        Self {
            addr: Some(RedisConfig::ConnAddr(conn_addr)),
            channels: Vec::new(),
            patterns: Vec::new(),
            retry: RetryAsync::new(),
        }
    }

    /// Creates a new `RedisPubSubSubscriberAsync` with a `ConnectionInfo`.
    ///
    /// # Arguments
    ///
    /// * `conn_info` - A Redis `ConnectionInfo`.
    ///
    /// # Returns
    ///
    /// A new instance of `RedisPubSubSubscriberAsync`.
    pub fn with_conn_info(conn_info: ConnectionInfo) -> Self {
        Self {
            addr: Some(RedisConfig::ConnInfo(conn_info)),
            channels: Vec::new(),
            patterns: Vec::new(),
            retry: RetryAsync::new(),
        }
    }

    /// Sets the retry configuration for the subscriber.
    ///
    /// # Arguments
    ///
    /// * `max_count` - The maximum number of retry attempts.
    /// * `init_delay_ms` - The initial delay between retries in milliseconds.
    /// * `max_delay_ms` - The maximum delay between retries in milliseconds.
    pub fn set_retry(&mut self, max_count: u32, init_delay_ms: u64, max_delay_ms: u64) {
        self.retry = RetryAsync::with_params(max_count, init_delay_ms, max_delay_ms);
    }

    /// Adds a channel to subscribe to.
    ///
    /// # Arguments
    ///
    /// * `channels` - The channel to subscribe to.
    pub fn subscribe(&mut self, channel: A) {
        self.channels.push(channel);
    }

    /// Adds a pattern to subscribe to.
    ///
    /// # Arguments
    ///
    /// * `patterns` - The pattern to subscribe to.
    pub fn psubscribe(&mut self, pattern: A) {
        self.patterns.push(pattern);
    }

    /// Starts receiving messages asynchronously and calls the provided callback for each message.
    ///
    /// This method continues until the callback returns `ControlFlow::Break`.
    ///
    /// # Arguments
    ///
    /// * `f` - A callback function that takes a `redis::Msg` and returns a `Future` that resolves to a `redis::ControlFlow`.
    ///
    /// # Returns
    ///
    /// A result containing the value returned by `ControlFlow::Break`, or an error.
    pub async fn receive_async<F, Fut, U>(mut self, mut f: F) -> errs::Result<U>
    where
        F: FnMut(Msg) -> Fut,
        Fut: Future<Output = ControlFlow<U>>,
    {
        let addr = self.addr.take().ok_or_else(|| {
            errs::Err::new(RedisPubSubSubscriberErrorAsync::AddressAlreadyConsumed)
        })?;
        let client = match addr {
            RedisConfig::String(addr) => Client::open(addr.as_str()).map_err(|e| {
                errs::Err::with_source(
                    RedisPubSubSubscriberErrorAsync::FailToOpenClientOfAddr { addr },
                    e,
                )
            })?,
            RedisConfig::ConnAddr(conn_addr) => Client::open(conn_addr.clone()).map_err(|e| {
                errs::Err::with_source(
                    RedisPubSubSubscriberErrorAsync::FailToOpenClientOfConnAddr { conn_addr },
                    e,
                )
            })?,
            RedisConfig::ConnInfo(conn_info) => Client::open(conn_info.clone()).map_err(|e| {
                errs::Err::with_source(
                    RedisPubSubSubscriberErrorAsync::FailToOpenClientOfConnInfo { conn_info },
                    e,
                )
            })?,
        };

        loop {
            let pubsub: PubSub = match client.get_async_pubsub().await {
                Ok(pubsub) => pubsub,
                Err(e) => {
                    if self.retry.wait_with_backoff_async().await {
                        continue;
                    }
                    return Err(errs::Err::with_source(
                        RedisPubSubSubscriberErrorAsync::FailToGetAsyncPubSub {
                            conn_info: client.get_connection_info().clone(),
                        },
                        e,
                    ));
                }
            };
            let (mut sink, mut stream) = pubsub.split();

            for c in self.channels.iter() {
                sink.subscribe(c).await.map_err(|e| {
                    errs::Err::with_source(
                        RedisPubSubSubscriberErrorAsync::FailToSubscribeToChannels {
                            conn_info: client.get_connection_info().clone(),
                        },
                        e,
                    )
                })?;
            }

            for p in self.patterns.iter() {
                sink.psubscribe(p).await.map_err(|e| {
                    errs::Err::with_source(
                        RedisPubSubSubscriberErrorAsync::FailToSubscribeToChannelsWithPatterns {
                            conn_info: client.get_connection_info().clone(),
                        },
                        e,
                    )
                })?;
            }

            loop {
                match stream.next().await {
                    Some(msg) => {
                        self.retry.reset();
                        if let ControlFlow::Break(value) = f(msg).await {
                            return Ok(value);
                        }
                    }
                    None => {
                        if self.retry.wait_with_backoff_async().await {
                            continue;
                        }
                        return Err(errs::Err::new(
                            RedisPubSubSubscriberErrorAsync::FailToGetMessage {
                                conn_info: client.get_connection_info().clone(),
                            },
                        ));
                    }
                }
            }
        }
    }
}

#[cfg(test)]
mod unit_tests {
    use super::*;
    use crate::RedisDataSrcAsync;
    use redis::AsyncTypedCommands;
    use sabi::tokio::{AsyncGroup, DataSrc};
    use url::Url;

    async fn publish_async(s: &str) {
        let s = s.to_string();
        let _ = tokio::spawn(async move {
            let mut ds = RedisDataSrcAsync::new("redis://127.0.0.1:6379/8");
            let mut ag = AsyncGroup::new();
            ds.setup_async(&mut ag).await.unwrap();
            let mut dc = ds.create_data_conn_async().await.unwrap();
            let conn = dc.get_connection();
            tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
            conn.publish("channel-1", s).await.unwrap();
        });
    }

    mod test_new {
        use super::*;

        #[tokio::test]
        async fn addr_is_str_and_ok() {
            publish_async("Hello").await;

            let mut subscriber = RedisPubSubSubscriberAsync::new("redis://127.0.0.1/0");

            subscriber.subscribe("channel-1");
            subscriber
                .receive_async(async |msg| {
                    let payload: String = msg.get_payload().unwrap();
                    assert_eq!(payload, "Hello");
                    ControlFlow::Break(1)
                })
                .await
                .unwrap();
        }

        #[tokio::test]
        async fn addr_is_str_and_fail() {
            let mut subscriber = RedisPubSubSubscriberAsync::new("xxxx");

            subscriber.set_retry(1, 0, 0);
            subscriber.subscribe("channel-1");
            let Err(err): errs::Result<i32> = subscriber.receive_async(async |_msg| panic!()).await
            else {
                panic!();
            };
            let Ok(RedisPubSubSubscriberErrorAsync::FailToOpenClientOfAddr { addr }) =
                err.reason::<RedisPubSubSubscriberErrorAsync>()
            else {
                panic!();
            };
            assert_eq!(addr, "xxxx");
            let Some(src) = err.source() else {
                panic!();
            };
            assert_eq!(
                format!("{src:?}"),
                "Redis URL did not parse - InvalidClientConfig"
            );
        }

        #[tokio::test]
        async fn addr_is_string_and_ok() {
            publish_async("Hello").await;

            let mut subscriber = RedisPubSubSubscriberAsync::new("redis://127.0.0.1/0".to_string());

            subscriber.subscribe("channel-1");
            subscriber
                .receive_async(async |msg| {
                    let payload: String = msg.get_payload().unwrap();
                    assert_eq!(payload, "Hello");
                    ControlFlow::Break(1)
                })
                .await
                .unwrap();
        }

        #[tokio::test]
        async fn addr_is_string_and_fail() {
            let mut subscriber = RedisPubSubSubscriberAsync::new("xxxx".to_string());

            subscriber.set_retry(1, 0, 0);
            subscriber.subscribe("channel-1");
            let Err(err): errs::Result<i32> = subscriber.receive_async(async |_msg| panic!()).await
            else {
                panic!();
            };
            let Ok(RedisPubSubSubscriberErrorAsync::FailToOpenClientOfAddr { addr }) =
                err.reason::<RedisPubSubSubscriberErrorAsync>()
            else {
                panic!();
            };
            assert_eq!(addr, "xxxx");
            let Some(src) = err.source() else {
                panic!();
            };
            assert_eq!(
                format!("{src:?}"),
                "Redis URL did not parse - InvalidClientConfig"
            );
        }

        #[tokio::test]
        async fn addr_is_url_and_ok() {
            publish_async("Hello").await;

            let Ok(url) = Url::parse("redis://127.0.0.1:6379/0") else {
                panic!("bad url");
            };
            let mut subscriber = RedisPubSubSubscriberAsync::new(url);

            subscriber.subscribe("channel-1");
            subscriber
                .receive_async(async |msg| {
                    let payload: String = msg.get_payload().unwrap();
                    assert_eq!(payload, "Hello");
                    ControlFlow::Break(1)
                })
                .await
                .unwrap();
        }

        #[tokio::test]
        async fn addr_is_url_and_fail() {
            let Ok(url) = Url::parse("redis://xxxx") else {
                panic!("bad url");
            };
            let mut subscriber = RedisPubSubSubscriberAsync::new(url);

            subscriber.set_retry(1, 0, 0);
            subscriber.subscribe("channel-1");
            let Err(err): errs::Result<i32> = subscriber.receive_async(async |_msg| panic!()).await
            else {
                panic!();
            };
            let Ok(RedisPubSubSubscriberErrorAsync::FailToGetAsyncPubSub { conn_info }) =
                err.reason::<RedisPubSubSubscriberErrorAsync>()
            else {
                panic!();
            };
            #[cfg(target_os = "linux")]
            assert_eq!(format!("{conn_info:?}"), "ConnectionInfo { addr: Tcp(\"xxxx\", 6379), tcp_settings: TcpSettings { nodelay: false, keepalive: None, user_timeout: None }, redis: RedisConnectionInfo { db: 0, username: None, password: None, protocol: RESP2, skip_set_lib_name: false, lib_name: None, lib_ver: None } }");
            #[cfg(not(target_os = "linux"))]
            assert_eq!(format!("{conn_info:?}"), "ConnectionInfo { addr: Tcp(\"xxxx\", 6379), tcp_settings: TcpSettings { nodelay: false, keepalive: None }, redis: RedisConnectionInfo { db: 0, username: None, password: None, protocol: RESP2, skip_set_lib_name: false, lib_name: None, lib_ver: None } }");
            let Some(src) = err.source() else {
                panic!();
            };
            #[cfg(target_os = "linux")]
            assert_eq!(
                format!("{src:?}"),
                "failed to lookup address information: Temporary failure in name resolution"
            );
            #[cfg(not(target_os = "linux"))]
            assert_eq!(format!("{src:?}"), "failed to lookup address information: nodename nor servname provided, or not known");
        }
    }

    mod test_with_conn_addr {
        use super::*;

        #[tokio::test]
        async fn ok() {
            publish_async("Hello").await;

            let conn_addr = redis::ConnectionAddr::Tcp("127.0.0.1".to_string(), 6379);
            let mut subscriber = RedisPubSubSubscriberAsync::with_conn_addr(conn_addr);

            subscriber.subscribe("channel-1");
            subscriber
                .receive_async(async |msg| {
                    let payload: String = msg.get_payload().unwrap();
                    assert_eq!(payload, "Hello");
                    ControlFlow::Break(1)
                })
                .await
                .unwrap();
        }

        #[tokio::test]
        async fn fail() {
            let conn_addr = redis::ConnectionAddr::Tcp("xxxx".to_string(), 6379);
            let mut subscriber = RedisPubSubSubscriberAsync::with_conn_addr(conn_addr);

            subscriber.set_retry(1, 0, 0);
            subscriber.subscribe("channel-1");
            let Err(err): errs::Result<i32> = subscriber.receive_async(async |_msg| panic!()).await
            else {
                panic!();
            };
            let Ok(RedisPubSubSubscriberErrorAsync::FailToGetAsyncPubSub { conn_info }) =
                err.reason::<RedisPubSubSubscriberErrorAsync>()
            else {
                panic!();
            };
            #[cfg(target_os = "linux")]
            assert_eq!(format!("{conn_info:?}"), "ConnectionInfo { addr: Tcp(\"xxxx\", 6379), tcp_settings: TcpSettings { nodelay: false, keepalive: None, user_timeout: None }, redis: RedisConnectionInfo { db: 0, username: None, password: None, protocol: RESP2, skip_set_lib_name: false, lib_name: None, lib_ver: None } }");
            #[cfg(not(target_os = "linux"))]
            assert_eq!(format!("{conn_info:?}"), "ConnectionInfo { addr: Tcp(\"xxxx\", 6379), tcp_settings: TcpSettings { nodelay: false, keepalive: None }, redis: RedisConnectionInfo { db: 0, username: None, password: None, protocol: RESP2, skip_set_lib_name: false, lib_name: None, lib_ver: None } }");
            let Some(src) = err.source() else {
                panic!();
            };
            #[cfg(target_os = "linux")]
            assert_eq!(
                format!("{src:?}"),
                "failed to lookup address information: Temporary failure in name resolution"
            );
            #[cfg(not(target_os = "linux"))]
            assert_eq!(format!("{src:?}"), "failed to lookup address information: nodename nor servname provided, or not known");
        }
    }

    mod test_with_conn_info {
        use super::*;
        use redis::IntoConnectionInfo;

        #[tokio::test]
        async fn ok() {
            publish_async("Hello").await;

            let conn_info = "redis://127.0.0.1:6379/0".into_connection_info().unwrap();
            let mut subscriber = RedisPubSubSubscriberAsync::with_conn_info(conn_info);

            subscriber.subscribe("channel-1");
            subscriber
                .receive_async(async |msg| {
                    let payload: String = msg.get_payload().unwrap();
                    assert_eq!(payload, "Hello");
                    ControlFlow::Break(1)
                })
                .await
                .unwrap();
        }

        #[tokio::test]
        async fn fail() {
            let conn_info = "redis://xxxx".into_connection_info().unwrap();
            let mut subscriber = RedisPubSubSubscriberAsync::with_conn_info(conn_info);

            subscriber.set_retry(1, 0, 0);
            subscriber.subscribe("channel-1");
            let Err(err): errs::Result<i32> = subscriber.receive_async(async |_msg| panic!()).await
            else {
                panic!();
            };
            let Ok(RedisPubSubSubscriberErrorAsync::FailToGetAsyncPubSub { conn_info }) =
                err.reason::<RedisPubSubSubscriberErrorAsync>()
            else {
                panic!();
            };
            #[cfg(target_os = "linux")]
            assert_eq!(format!("{conn_info:?}"), "ConnectionInfo { addr: Tcp(\"xxxx\", 6379), tcp_settings: TcpSettings { nodelay: false, keepalive: None, user_timeout: None }, redis: RedisConnectionInfo { db: 0, username: None, password: None, protocol: RESP2, skip_set_lib_name: false, lib_name: None, lib_ver: None } }");
            #[cfg(not(target_os = "linux"))]
            assert_eq!(format!("{conn_info:?}"), "ConnectionInfo { addr: Tcp(\"xxxx\", 6379), tcp_settings: TcpSettings { nodelay: false, keepalive: None }, redis: RedisConnectionInfo { db: 0, username: None, password: None, protocol: RESP2, skip_set_lib_name: false, lib_name: None, lib_ver: None } }");
            let Some(src) = err.source() else {
                panic!();
            };
            #[cfg(target_os = "linux")]
            assert_eq!(
                format!("{src:?}"),
                "failed to lookup address information: Temporary failure in name resolution"
            );
            #[cfg(not(target_os = "linux"))]
            assert_eq!(format!("{src:?}"), "failed to lookup address information: nodename nor servname provided, or not known");
        }
    }

    mod subscribe {
        use super::*;
        use redis::IntoConnectionInfo;

        #[tokio::test]
        async fn ok() {
            publish_async("Hello").await;

            let conn_info = "redis://127.0.0.1:6379/0".into_connection_info().unwrap();
            let mut subscriber = RedisPubSubSubscriberAsync::with_conn_info(conn_info);

            subscriber.subscribe("channel-1");
            subscriber
                .receive_async(async |msg| {
                    let payload: String = msg.get_payload().unwrap();
                    assert_eq!(payload, "Hello");
                    ControlFlow::Break(1)
                })
                .await
                .unwrap();
        }
    }

    mod psubscribe {
        use super::*;
        use redis::IntoConnectionInfo;

        #[tokio::test]
        async fn ok() {
            publish_async("Hello").await;

            let conn_info = "redis://127.0.0.1:6379/0".into_connection_info().unwrap();
            let mut subscriber = RedisPubSubSubscriberAsync::with_conn_info(conn_info);

            subscriber.psubscribe("channel-*");
            subscriber
                .receive_async(async |msg| {
                    let payload: String = msg.get_payload().unwrap();
                    assert_eq!(payload, "Hello");
                    ControlFlow::Break(1)
                })
                .await
                .unwrap();
        }
    }
}