tcpclient 2.1.2

//! # tcpclient
//!
//! 基于 [`aqueue`] actor 模型的异步 TCP 客户端。
//! Asynchronous TCP client built on the [`aqueue`] actor model.
//!
//! ## 核心设计 / Core Design
//!
//! 将 Tokio 的 `TcpStream` 拆分为读写两半：
//! - **写半（WriteHalf）** 由 [`TcpClient`] 持有，所有写操作通过 [`aqueue::Actor`]
//!   串行化，确保单写入者语义，无需外部锁。
//! - **读半（ReadHalf）** 交由用户提供的异步闭包处理，运行在独立 Tokio 任务中。
//!
//! The library splits a Tokio `TcpStream` into read/write halves:
//! - The **write half** is owned by [`TcpClient`], with all writes serialized
//!   through an [`aqueue::Actor`], guaranteeing single-writer semantics with no locks.
//! - The **read half** is handed to a user-supplied async closure running in a spawned Tokio task.
//!
//! ## 使用示例 / Example
//!
//! ```rust,ignore
//! use tcpclient::{TcpClient, SocketClientTrait};
//! use tokio::io::AsyncReadExt;
//!
//! #[tokio::main]
//! async fn main() -> Result<(), Box<dyn std::error::Error>> {
//!     // 连接到 echo 服务器 / Connect to echo server
//!     let client = TcpClient::connect(
//!         "127.0.0.1:5555",
//!         async move |_, client, mut reader| {
//!             let mut buff = [0u8; 7];
//!             while let Ok(len) = reader.read_exact(&mut buff).await {
//!                 println!("{}", std::str::from_utf8(&buff[..len])?);
//!                 client.send(&buff[..len]).await?;
//!             }
//!             Ok(true) // true = 退出时断开 / disconnect on exit
//!         },
//!         (), // token（上下文数据 / context data）
//!     ).await?;
//!
//!     // 发送数据 / Send data
//!     client.send_all_ref(b"1234567").await?;
//!
//!     // 断开连接 / Disconnect
//!     client.disconnect().await?;
//!     Ok(())
//! }
//! ```

pub mod error;

use aqueue::Actor;
use error::Result;
use log::*;
use std::borrow::Cow;
use std::future::Future;
use std::net::SocketAddr;
use std::ops::Deref;
use std::sync::Arc;
use std::time::Duration;
use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt, ReadHalf, WriteHalf};
use tokio::net::{TcpStream, ToSocketAddrs};

// ============================================================================
// TcpClient — 内部客户端实现 / Internal client implementation
// ============================================================================

/// TCP 客户端，持有流的写半、对端地址和连接状态。
/// TCP client holding the write half, peer address, and connection state.
///
/// 该类型**不应**被用户直接使用；用户通过
/// [`Arc<Actor<TcpClient<T>>>`] 和 [`SocketClientTrait`] 进行操作。
///
/// This type is **not** intended for direct use; users interact via
/// [`Arc<Actor<TcpClient<T>>>`] and the [`SocketClientTrait`].
pub struct TcpClient<T> {
    /// 是否已断开连接 / Whether already disconnected.
    /// 由 `disconnect()` 设置为 `true`，防止重复关闭。
    /// Set to `true` by `disconnect()` to prevent duplicate shutdowns.
    disconnect: bool,
    /// 流的写半 / Write half of the tokio stream.
    /// 所有写操作必须通过 `Actor::inner_call` 串行访问。
    /// All writes must be serialized through `Actor::inner_call`.
    sender: WriteHalf<T>,
    /// 对端地址 / Peer socket address.
    /// 存储用于诊断日志和错误消息。
    /// Stored for diagnostic logging and error messages.
    peer_addr: SocketAddr,
}

// ============================================================================
// TcpClient<TcpStream> — 纯 TCP 连接 / Plain TCP connection
// ============================================================================

impl TcpClient<TcpStream> {
    /// 连接到一个 TCP 地址并初始化客户端。
    /// Connect to a TCP address and initialize the client.
    ///
    /// 这是最常用的入口：解析地址 → 建立 TCP 连接 → 拆分流 →
    /// 创建 Actor → 在 Tokio 任务中运行用户提供的读闭包。
    ///
    /// This is the primary entry point: resolve addr → establish TCP connection →
    /// split the stream → create Actor → run user's read closure in a Tokio task.
    ///
    /// # 参数 / Parameters
    ///
    /// - `addr`: 目标地址，实现 `ToSocketAddrs`。
    /// - `input`: 用户提供的读循环闭包。参数为 `(token, client_handle, read_half)`，
    ///   返回 `Result<bool>`：`Ok(true)` 退出时断开连接，`Ok(false)` 正常关闭。
    /// - `token`: 传递给 `input` 闭包的任意上下文数据。
    ///
    /// # 返回 / Returns
    ///
    /// `Arc<Actor<TcpClient<TcpStream>>>` — 客户端的共享句柄，可 clone 多任务发送。
    #[inline]
    pub async fn connect<
        Addr: ToSocketAddrs,
        Fut: Future<Output = std::result::Result<bool, E>> + Send + 'static,
        E: std::fmt::Display + Send + 'static,
        Token: Send + 'static,
    >(
        addr: Addr,
        input: impl FnOnce(Token, Arc<Actor<TcpClient<TcpStream>>>, ReadHalf<TcpStream>) -> Fut
            + Send
            + 'static,
        token: Token,
    ) -> Result<Arc<Actor<TcpClient<TcpStream>>>> {
        // 建立 TCP 连接，获取对端地址
        let stream = TcpStream::connect(addr).await?;
        let target = stream.peer_addr()?;
        Self::init(input, token, stream, target)
    }

    /// 带超时的 TCP 连接。
    /// Connect with a timeout on the TCP handshake.
    ///
    /// 用 [`tokio::time::timeout`] 包装 [`TcpStream::connect`]。
    /// 如果在 `duration` 内未完成连接，返回 `IOError`（`TimedOut` 类型）。
    ///
    /// Wraps [`TcpStream::connect`] with [`tokio::time::timeout`].
    /// Returns an `IOError` with `TimedOut` kind if the connection
    /// does not complete within `duration`.
    ///
    /// # 参数 / Parameters
    ///
    /// - `addr`: 目标地址 / Target address.
    /// - `duration`: 连接超时时间 / Connection timeout duration.
    /// - `input`: 用户读闭包（同 [`connect`](Self::connect)）/ User read closure (same as [`connect`](Self::connect)).
    /// - `token`: 上下文数据 / Context data.
    #[inline]
    pub async fn connect_with_timeout<
        Addr: ToSocketAddrs,
        Fut: Future<Output = std::result::Result<bool, E>> + Send + 'static,
        E: std::fmt::Display + Send + 'static,
        Token: Send + 'static,
    >(
        addr: Addr,
        duration: Duration,
        input: impl FnOnce(Token, Arc<Actor<TcpClient<TcpStream>>>, ReadHalf<TcpStream>) -> Fut
            + Send
            + 'static,
        token: Token,
    ) -> Result<Arc<Actor<TcpClient<TcpStream>>>> {
        let stream = tokio::time::timeout(duration, TcpStream::connect(addr))
            .await
            .map_err(|_| {
                std::io::Error::new(std::io::ErrorKind::TimedOut, "connection timed out")
            })??;
        let target = stream.peer_addr()?;
        Self::init(input, token, stream, target)
    }
}

// ============================================================================
// TcpClient<T> — 通用流类型（支持 TLS 等） / Generic stream type (TLS, etc.)
// ============================================================================

impl<T> TcpClient<T>
where
    T: AsyncRead + AsyncWrite + Send + 'static,
{
    /// 连接到地址并使用自定义流初始化器包装原始 TCP 流。
    /// Connect to an address and wrap the raw TCP stream with a custom initializer.
    ///
    /// 用于 TLS 等场景：先建立 TCP 连接，再通过 `stream_init` 将
    /// `TcpStream` 升级为 `TlsStream<TcpStream>` 等类型。
    ///
    /// Use for TLS etc.: first establish a TCP connection, then upgrade the
    /// raw `TcpStream` via `stream_init` (e.g. TLS handshake).
    ///
    /// # 参数 / Parameters
    ///
    /// - `addr`: 目标地址。
    /// - `stream_init`: 将 `TcpStream` 转换为目标流类型（如 TLS 握手）。
    /// - `input`: 用户读闭包，与 [`connect`](Self::connect) 相同语义。
    /// - `token`: 传递给 `input` 的上下文数据。
    #[inline]
    pub async fn connect_stream_type<
        Addr: ToSocketAddrs,
        Fut: Future<Output = std::result::Result<bool, E>> + Send + 'static,
        E: std::fmt::Display + Send + 'static,
        StreamFut: Future<Output = anyhow::Result<T>> + Send + 'static,
        Token: Send + 'static,
    >(
        addr: Addr,
        stream_init: impl FnOnce(TcpStream) -> StreamFut + Send + 'static,
        input: impl FnOnce(Token, Arc<Actor<TcpClient<T>>>, ReadHalf<T>) -> Fut + Send + 'static,
        token: Token,
    ) -> Result<Arc<Actor<TcpClient<T>>>> {
        // 建立 TCP 连接
        let stream = TcpStream::connect(addr).await?;
        let target = stream.peer_addr()?;
        // 通过用户提供的初始化器包装流（例如 TLS 握手）
        let stream = stream_init(stream).await?;
        Self::init(input, token, stream, target)
    }

    /// 带超时的通用流类型连接（仅对 TCP 握手阶段计时）。
    /// Connect with a custom stream initializer and timeout on the TCP handshake.
    ///
    /// 仅对 TCP `connect` 阶段计时；`stream_init` 阶段（如 TLS 握手）不计时。
    /// 如需对整个连接过程计时，可用 `tokio::time::timeout` 包装整个调用。
    ///
    /// Only the TCP `connect` phase is timed; the `stream_init` phase
    /// (e.g., TLS handshake) runs without an additional timeout.
    /// Wrap the entire call in `tokio::time::timeout` if both phases need a deadline.
    #[inline]
    pub async fn connect_stream_type_with_timeout<
        Addr: ToSocketAddrs,
        Fut: Future<Output = std::result::Result<bool, E>> + Send + 'static,
        E: std::fmt::Display + Send + 'static,
        StreamFut: Future<Output = anyhow::Result<T>> + Send + 'static,
        Token: Send + 'static,
    >(
        addr: Addr,
        duration: Duration,
        stream_init: impl FnOnce(TcpStream) -> StreamFut + Send + 'static,
        input: impl FnOnce(Token, Arc<Actor<TcpClient<T>>>, ReadHalf<T>) -> Fut + Send + 'static,
        token: Token,
    ) -> Result<Arc<Actor<TcpClient<T>>>> {
        let stream = tokio::time::timeout(duration, TcpStream::connect(addr))
            .await
            .map_err(|_| {
                std::io::Error::new(std::io::ErrorKind::TimedOut, "connection timed out")
            })??;
        let target = stream.peer_addr()?;
        let stream = stream_init(stream).await?;
        Self::init(input, token, stream, target)
    }

    /// 内部初始化方法：拆分流 → 创建 Actor → 启动读任务。
    /// Internal init: split stream → create Actor → spawn reader task.
    ///
    /// 这是 [`connect`](Self::connect) 和 [`connect_stream_type`](Self::connect_stream_type)
    /// 的公共后段逻辑。
    /// Shared tail of both `connect` and `connect_stream_type`.
    #[inline]
    fn init<Fut, E, Token>(
        f: impl FnOnce(Token, Arc<Actor<TcpClient<T>>>, ReadHalf<T>) -> Fut + Send + 'static,
        token: Token,
        stream: T,
        target: SocketAddr,
    ) -> Result<Arc<Actor<TcpClient<T>>>>
    where
        Fut: Future<Output = std::result::Result<bool, E>> + Send + 'static,
        E: std::fmt::Display + Send + 'static,
        Token: Send + 'static,
    {
        // 将流拆分为读写两半
        let (reader, sender) = tokio::io::split(stream);

        // 创建 Actor 包装的 TcpClient，持有写半
        let client = Arc::new(Actor::new(TcpClient {
            disconnect: false,
            sender,
            peer_addr: target,
        }));

        // 克隆句柄供 reader task 使用
        let read_client = client.clone();

        // 在独立 Tokio 任务中运行用户提供的读闭包
        tokio::spawn(async move {
            let disconnect_client = read_client.clone();

            // 运行用户读循环：Ok(true) 需要断开，Ok(false) 正常关闭，Err 强制断开
            let need_disconnect = f(token, read_client, reader).await.unwrap_or_else(|err| {
                error!("reader error:{}", err);
                true // 读出错时强制断开连接 / Force disconnect on read error
            });

            if need_disconnect {
                if let Err(er) = disconnect_client.disconnect().await {
                    error!("disconnect to{} err:{}", target, er);
                } else {
                    debug!("disconnect to {}", target);
                }
            } else {
                debug!("{} reader is close", target);
            }
        });

        Ok(client)
    }

    // ------------------------------------------------------------------
    // 内部辅助方法 / Internal helpers
    // ------------------------------------------------------------------

    /// 检查连接状态，已断开时返回 `SendError`。
    /// Check connection state; return `SendError` if already disconnected.
    #[inline]
    fn ensure_connected(&self) -> Result<()> {
        if self.disconnect {
            Err(error::Error::SendError(Cow::Owned(format!(
                "Disconnect, peer:{}",
                self.peer_addr
            ))))
        } else {
            Ok(())
        }
    }

    // ------------------------------------------------------------------
    // 内部方法 — 由 Actor 内部调用 / Internal methods — called inside Actor
    // ------------------------------------------------------------------

    /// 断开连接（关闭写半）。
    /// Disconnect by shutting down the write half.
    ///
    /// 幂等操作：首次调用执行 `shutdown()`，后续调用直接返回成功。
    /// 如果 `shutdown()` 失败，`disconnect` 标志保持在 `false`，允许重试。
    ///
    /// Idempotent: the first call performs `shutdown()`, subsequent calls are no-ops.
    /// If `shutdown()` fails, the flag stays `false`, allowing a retry.
    #[inline]
    async fn disconnect(&mut self) -> Result<()> {
        if !self.disconnect {
            self.sender.shutdown().await?;
            self.disconnect = true;
        }
        Ok(())
    }

    /// 发送字节数据（尽力而为，不保证全部写入）。
    /// Send bytes (best-effort; may not write all data).
    ///
    /// 返回实际写入的字节数。如需保证全部写入，请使用 `send_all_ref` 或 `send_all`。
    /// Returns the number of bytes written. For guaranteed full writes, use `send_all_ref` or `send_all`.
    #[inline]
    async fn send(&mut self, buff: &[u8]) -> Result<usize> {
        self.ensure_connected()?;
        Ok(self.sender.write(buff).await?)
    }

    /// 发送全部字节并刷新（保证完整写入）。
    /// Send all bytes and flush (guarantees complete write).
    ///
    /// 内部方法，通过 [`SocketClientTrait::send_all`] 和
    /// [`SocketClientTrait::send_all_ref`] 对外暴露。
    ///
    /// Internal method, exposed via [`SocketClientTrait::send_all`] and
    /// [`SocketClientTrait::send_all_ref`].
    #[inline]
    async fn send_all(&mut self, buff: &[u8]) -> Result<()> {
        self.ensure_connected()?;
        self.sender.write_all(buff).await?;
        Ok(self.sender.flush().await?)
    }

    /// 刷新写缓冲区，确保所有已缓冲的数据被发送到对端。
    /// Flush the write buffer, ensuring all buffered data reaches the peer.
    #[inline]
    async fn flush(&mut self) -> Result<()> {
        self.ensure_connected()?;
        Ok(self.sender.flush().await?)
    }
}

// ============================================================================
// SocketClientTrait — 面向 Actor 包装的公共 API
// Public trait providing the actor-based API
// ============================================================================

/// TCP 客户端操作的公共 trait，实现于 [`Arc<Actor<TcpClient<T>>>`]。
/// Public trait for TCP client operations, implemented on [`Arc<Actor<TcpClient<T>>>`].
///
/// 每个方法通过 [`Actor::inner_call`] 将操作排队，确保对
/// [`TcpClient`] 内部状态的独占 `&mut` 访问。
///
/// Each method enqueues the operation through [`Actor::inner_call`], providing
/// exclusive `&mut` access to the inner [`TcpClient`] — thread safety without locks.
///
/// # 缓冲区类型说明 / Buffer type notes
///
/// - 所有权版本（`send` / `send_all`）接受 `B: Deref<Target=[u8]>`，
///   可传递 `Vec<u8>`、`Box<[u8]>`、`String` 等。
/// - 引用版本（`send_ref` / `send_all_ref`）接受 `&[u8]`，并要求非空。
///
/// - Owned variants (`send` / `send_all`) accept `B: Deref<Target=[u8]>`,
///   allowing `Vec<u8>`, `Box<[u8]>`, `String`, etc.
/// - Borrowed variants (`send_ref` / `send_all_ref`) accept `&[u8]` with a non-empty assertion.
pub trait SocketClientTrait {
    /// 发送数据（所有权版本）。
    /// Send data (owned buffer via `Deref<Target=[u8]>`).
    #[must_use]
    fn send<B: Deref<Target = [u8]> + Send + 'static>(
        &self,
        buff: B,
    ) -> impl Future<Output = Result<usize>>;

    /// 发送全部数据并刷新（所有权版本）。
    /// Send all data and flush (owned buffer version).
    fn send_all<B: Deref<Target = [u8]> + Send + 'static>(
        &self,
        buff: B,
    ) -> impl Future<Output = Result<()>>;

    /// 发送数据（引用版本，要求非空）。
    /// Send data (borrowed buffer `&[u8]`, requires non-empty).
    #[must_use]
    fn send_ref(&self, buff: &[u8]) -> impl Future<Output = Result<usize>>;

    /// 发送全部数据并刷新（引用版本，要求非空）。
    /// Send all data and flush (borrowed buffer version, requires non-empty).
    fn send_all_ref(&self, buff: &[u8]) -> impl Future<Output = Result<()>>;

    /// 刷新写缓冲区。
    /// Flush the write buffer.
    fn flush(&self) -> impl Future<Output = Result<()>>;

    /// 断开连接（幂等，可安全重复调用）。
    /// Disconnect (idempotent; safe to call multiple times).
    fn disconnect(&self) -> impl Future<Output = Result<()>>;
}

// ============================================================================
// SocketClientTrait 的 Actor<TcpClient<T>> 实现
// ============================================================================

impl<T> SocketClientTrait for Actor<TcpClient<T>>
where
    T: AsyncRead + AsyncWrite + Send + 'static,
{
    /// 通过 Actor 发送数据。
    /// `inner_call` 获取 `&mut TcpClient<T>` 后调用内部 `send`。
    #[inline]
    async fn send<B: Deref<Target = [u8]> + Send + 'static>(&self, buff: B) -> Result<usize> {
        self.inner_call(|inner| async move { inner.get_mut().send(&buff).await })
            .await
    }

    /// 通过 Actor 发送全部数据并刷新。
    #[inline]
    async fn send_all<B: Deref<Target = [u8]> + Send + 'static>(&self, buff: B) -> Result<()> {
        self.inner_call(|inner| async move { inner.get_mut().send_all(&buff).await })
            .await
    }

    /// 通过 Actor 发送引用数据。
    /// 额外增加非空校验，空缓冲区返回 `SendError`。
    #[inline]
    async fn send_ref(&self, buff: &[u8]) -> Result<usize> {
        if buff.is_empty() {
            return Err(error::Error::SendError(Cow::Borrowed("send buff is none")));
        }
        self.inner_call(|inner| async move { inner.get_mut().send(buff).await })
            .await
    }

    /// 通过 Actor 发送全部引用数据并刷新。
    /// 额外增加非空校验，空缓冲区返回 `SendError`。
    #[inline]
    async fn send_all_ref(&self, buff: &[u8]) -> Result<()> {
        if buff.is_empty() {
            return Err(error::Error::SendError(Cow::Borrowed("send buff is none")));
        }
        self.inner_call(|inner| async move { inner.get_mut().send_all(buff).await })
            .await
    }

    /// 通过 Actor 刷新写缓冲区。
    #[inline]
    async fn flush(&self) -> Result<()> {
        self.inner_call(|inner| async move { inner.get_mut().flush().await })
            .await
    }

    /// 通过 Actor 断开连接。
    #[inline]
    async fn disconnect(&self) -> Result<()> {
        self.inner_call(|inner| async move { inner.get_mut().disconnect().await })
            .await
    }
}

// ============================================================================
// Tests
// ============================================================================

#[cfg(test)]
mod tests {
    use super::*;
    use tokio::net::TcpListener;

    /// 测试 SendError 的 Display 实现正确。
    /// Verify SendError Display implementation is correct.
    #[test]
    fn test_send_error_display() {
        let err = error::Error::SendError(Cow::Borrowed("test message"));
        assert_eq!(err.to_string(), "test message");
    }

    /// 测试 std::io::Error 到 crate Error 的自动转换。
    /// Verify auto-conversion from std::io::Error via #[from].
    #[test]
    fn test_io_error_conversion() {
        let io_err = std::io::Error::new(std::io::ErrorKind::ConnectionRefused, "refused");
        let err: error::Error = io_err.into();
        assert!(matches!(err, error::Error::IOError(_)));
    }

    /// 集成测试：本地 echo 服务器 → 连接 → 发送 → 验证回显 → 断开。
    /// Integration test: local echo server → connect → send → verify echo → disconnect.
    #[tokio::test]
    async fn test_echo_client() {
        // 在随机端口上启动 echo 服务器 / Start echo server on random port
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();

        tokio::spawn(async move {
            let (socket, _) = listener.accept().await.unwrap();
            let (mut reader, mut writer) = tokio::io::split(socket);
            let _ = tokio::io::copy(&mut reader, &mut writer).await;
        });

        // 连接到 echo 服务器 / Connect to echo server
        let client = TcpClient::connect(
            addr,
            async move |_, _client, _reader| {
                // 读任务不做任何事，由测试侧通过 send_all 和 disconnect 控制
                // Reader does nothing; the test side controls via send_all and disconnect
                Ok::<bool, Box<dyn std::error::Error + Send>>(false)
            },
            (),
        )
        .await
        .unwrap();

        // 发送数据并确保完整写入 / Send data and ensure complete write
        client.send_all_ref(b"hello echo").await.unwrap();

        // 断开连接 / Disconnect
        client.disconnect().await.unwrap();
    }

    /// 测试连接超时：对不可达地址使用极短超时必须返回错误。
    /// Test connect timeout: a very short timeout against an unreachable address must error.
    #[tokio::test]
    async fn test_connect_timeout() {
        // TEST-NET-1 地址通常不可达 / TEST-NET-1 is generally unreachable
        let result = TcpClient::connect_with_timeout(
            "192.0.2.1:9999",
            Duration::from_millis(100),
            async move |_, _client, _reader| Ok::<bool, Box<dyn std::error::Error + Send>>(false),
            (),
        )
        .await;
        assert!(result.is_err());
    }
}