tcpserver 2.2.2

use crate::error::Result;
use aqueue::Actor;
use std::io::ErrorKind;
use std::net::SocketAddr;
use std::ops::Deref;
use std::sync::Arc;
use tokio::io::WriteHalf;
use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt};

pub struct TCPPeer<T> {
    pub addr: SocketAddr,
    pub sender: Option<WriteHalf<T>>,
}

impl<T> TCPPeer<T>
where
    T: AsyncRead + AsyncWrite + Send + 'static,
{
    /// Create a new TCP peer wrapped in an Actor.
    ///
    /// 创建一个TCP PEER
    #[inline]
    pub fn new(addr: SocketAddr, sender: WriteHalf<T>) -> Arc<Actor<TCPPeer<T>>> {
        Arc::new(Actor::new(TCPPeer {
            addr,
            sender: Some(sender),
        }))
    }

    /// Returns `Ok(&mut sender)` or `Err(ConnectionReset)` if disconnected.
    #[inline]
    fn sender_mut(&mut self) -> Result<&mut WriteHalf<T>> {
        self.sender
            .as_mut()
            .ok_or_else(|| std::io::Error::from(ErrorKind::ConnectionReset).into())
    }

    /// Check whether the peer has been disconnected.
    ///
    /// 是否断线
    #[inline]
    pub fn is_disconnect(&self) -> bool {
        self.sender.is_none()
    }

    /// Send data to the peer. Returns the number of bytes written.
    ///
    /// 发送
    #[inline]
    pub async fn send(&mut self, buff: &[u8]) -> Result<usize> {
        Ok(self.sender_mut()?.write(buff).await?)
    }

    /// Send all data to the peer, then flush.
    ///
    /// 发送全部
    #[inline]
    pub async fn send_all(&mut self, buff: &[u8]) -> Result<()> {
        let sender = self.sender_mut()?;
        sender.write_all(buff).await?;
        sender.flush().await?;
        Ok(())
    }

    /// Flush the write buffer.
    ///
    /// flush
    #[inline]
    pub async fn flush(&mut self) -> Result<()> {
        self.sender_mut()?.flush().await?;
        Ok(())
    }

    /// Disconnect the peer by shutting down the write half.
    ///
    /// 掐线
    #[inline]
    pub async fn disconnect(&mut self) -> Result<()> {
        if let Some(mut sender) = self.sender.take() {
            sender.shutdown().await?;
        }
        Ok(())
    }
}

pub trait IPeer: Sync + Send {
    fn addr(&self) -> SocketAddr;
    fn is_disconnect(&self) -> impl std::future::Future<Output = Result<bool>>;
    fn send<B: Deref<Target = [u8]> + Send + Sync + 'static>(
        &self,
        buff: B,
    ) -> impl std::future::Future<Output = Result<usize>>;
    fn send_all<B: Deref<Target = [u8]> + Send + Sync + 'static>(
        &self,
        buff: B,
    ) -> impl std::future::Future<Output = Result<()>>;
    fn send_ref(&self, buff: &[u8]) -> impl std::future::Future<Output = Result<usize>>;
    fn send_all_ref(&self, buff: &[u8]) -> impl std::future::Future<Output = Result<()>>;
    fn flush(&self) -> impl std::future::Future<Output = Result<()>>;
    fn disconnect(&self) -> impl std::future::Future<Output = Result<()>>;
}

impl<T> IPeer for Actor<TCPPeer<T>>
where
    T: AsyncRead + AsyncWrite + Send + 'static,
{
    #[inline]
    fn addr(&self) -> SocketAddr {
        unsafe { self.deref_inner().addr }
    }

    #[inline]
    async fn is_disconnect(&self) -> Result<bool> {
        self.inner_call(|inner| async move { Ok(inner.get().is_disconnect()) })
            .await
    }

    #[inline]
    async fn send<B: Deref<Target = [u8]> + Send + Sync + 'static>(
        &self,
        buff: B,
    ) -> Result<usize> {
        self.inner_call(|inner| async move { inner.get_mut().send(&buff).await })
            .await
    }
    #[inline]
    async fn send_all<B: Deref<Target = [u8]> + Send + Sync + 'static>(
        &self,
        buff: B,
    ) -> Result<()> {
        self.inner_call(|inner| async move { inner.get_mut().send_all(&buff).await })
            .await
    }
    #[inline]
    async fn send_ref(&self, buff: &[u8]) -> Result<usize> {
        self.inner_call(|inner| async move { inner.get_mut().send(buff).await })
            .await
    }
    #[inline]
    async fn send_all_ref(&self, buff: &[u8]) -> Result<()> {
        self.inner_call(|inner| async move { inner.get_mut().send_all(buff).await })
            .await
    }

    #[inline]
    async fn flush(&self) -> Result<()> {
        self.inner_call(|inner| async move { inner.get_mut().flush().await })
            .await
    }

    #[inline]
    async fn disconnect(&self) -> Result<()> {
        self.inner_call(|inner| async move { inner.get_mut().disconnect().await })
            .await
    }
}