syslog_rs/a_sync/async_tokio/

async_socket.rs

/*-
 * syslog-rs - a syslog client translated from libc to rust
 * 
 * Copyright 2025 Aleksandr Morozov
 * 
 * The syslog-rs crate can be redistributed and/or modified
 * under the terms of either of the following licenses:
 *
 *   1. the Mozilla Public License Version 2.0 (the “MPL”) OR
 *                     
 *   2. EUROPEAN UNION PUBLIC LICENCE v. 1.2 EUPL © the European Union 2007, 2016
 */


use std::io::ErrorKind;
use std::net::Shutdown;

use tokio::net::{UnixDatagram};
use nix::errno::Errno;

use crate::a_sync::{AsyncSyslogTap, AsyncTap};



use crate::{map_error_os, throw_error_errno, throw_error_os, SyslogDestMsg, SyslogLocal, TapType};

use crate::common::*;
use crate::error::SyRes;


#[cfg(feature = "build_ext_tls")]
mod with_tls
{
    use std::io::ErrorKind;

    use tokio::{io::AsyncWriteExt, net::{TcpSocket, TcpStream}, time};
    use tokio_rustls::{TlsConnector};
    use tokio_rustls::client::TlsStream;

    use crate::{a_sync::{AsyncSyslogTap, AsyncTap}, error::SyRes, map_error, map_error_os, SyslogDestMsg, SyslogTls, TapType};

    impl AsyncSyslogTap<SyslogTls> for AsyncTap<TlsStream<TcpStream>, SyslogTls>
    {
        fn new(req_tap: SyslogTls) -> SyRes<Self>
        {
            let ret = 
                Self
                {
                    sock: None, 
                    tap_data: req_tap,
                    cur_tap_type: TapType::NetTcp,
                };

            return Ok(ret);
        }

        async 
        fn connectlog(&mut self) -> SyRes<()> 
        {

            let tcp_socket = 
                if self.tap_data.get_remote_addr().is_ipv4() == true
                {
                    TcpSocket::new_v4()
                }
                else
                {
                    TcpSocket::new_v6()
                }
                .map_err(|e|
                    map_error!("cannot create tcp socket '{}', error: {}", self.tap_data.get_remote_addr(), e)
                )?;

            // bind
            if self.tap_data.get_bind_addr().ip().is_unspecified() == false
            {
                tcp_socket
                    .bind(*self.tap_data.get_bind_addr())
                    .map_err(|e|
                        map_error!("cannot bind tcp socket '{}', error: {}", self.tap_data.get_remote_addr(), e)
                    )?;
            }
                
            let socket = 
                if let Some(c_timeout) = self.tap_data.get_get_conn_timeout()
                {
                    time::timeout(c_timeout, tcp_socket.connect(*self.tap_data.get_remote_addr()))
                        .await
                        .map_err(|e|
                            map_error!("can not connect to tcp '{}', error: {}", self.tap_data.get_remote_addr(), e)
                        )?
                }
                else
                {
                    tcp_socket
                        .connect(*self.tap_data.get_remote_addr())
                        .await
                }
                .map_err(|e|
                    map_error_os!(e, "cannot connect to tcp '{}', error: {}", self.tap_data.get_remote_addr(), e)
                )?;

            let connector = TlsConnector::from(self.tap_data.get_client_config());

            let stream = 
                connector
                    .connect(self.tap_data.get_serv_name(), socket)
                    .await
                    .map_err(|e|
                        map_error!("Remote server: '{}', TLS connector error: '{}'", self.tap_data.get_remote_addr(), e)
                    )?;

            // set socket
            self.sock = Some(stream);

            return Ok(());
        }

        #[inline]
        async 
        fn send(&mut self, msg: &[u8]) -> std::io::Result<usize> 
        {
            let sock = 
                self.sock
                    .as_mut()
                    .ok_or_else(||
                        std::io::Error::new(ErrorKind::NotConnected, "no connection")
                    )?;

            sock.write_all(msg).await?;
            sock.flush().await?;

            return Ok(msg.len());
        }

        async 
        fn disconnectlog(&mut self) -> std::io::Result<()> 
        {
            match self.sock.take()
            {
                Some(mut s) => 
                {
                    s.shutdown().await?;

                    drop(s);

                    Ok(())
                },
                None =>
                {
                    Ok(())
                }
            }
        }

        #[inline]
        fn is_connected(&self) -> bool 
        {
            return self.sock.is_some();
        }

        #[inline]
        fn get_type(&self) -> TapType 
        {
            return self.cur_tap_type;
        }

        #[inline]
        fn get_max_msg_size() -> usize
        {
            return crate::SyslogTls::get_max_msg_len();
        }

        #[inline]
        fn update_tap_data(&mut self, tap_data: SyslogTls)
        {
            self.tap_data = tap_data;
        }
    }
}


#[cfg(feature = "build_ext_net")]
mod with_net
{
    use std::io::ErrorKind;

    use tokio::{io::AsyncWriteExt, net::{TcpSocket, TcpStream, UdpSocket}, time};

    use crate::{a_sync::{AsyncSyslogTap, AsyncTap}, error::SyRes, map_error, map_error_os, SyslogDestMsg, SyslogNetTcp, SyslogNetUdp, TapType};

    impl AsyncSyslogTap<SyslogNetTcp> for AsyncTap<TcpStream, SyslogNetTcp>
    {
        fn new(req_tap: SyslogNetTcp) -> SyRes<Self>
        {
            let ret = 
                Self
                {
                    sock: None, 
                    tap_data: req_tap,
                    cur_tap_type: TapType::NetTcp,
                };

            return Ok(ret);
        }

        async 
        fn connectlog(&mut self) -> SyRes<()> 
        {
            let tcp_socket = 
                if self.tap_data.get_remote_addr().is_ipv4() == true
                {
                    TcpSocket::new_v4()
                }
                else
                {
                    TcpSocket::new_v6()
                }
                .map_err(|e|
                    map_error!("cannot create tcp socket '{}', error: {}", self.tap_data.get_remote_addr(), e)
                )?;

            // bind
            if self.tap_data.get_bind_addr().ip().is_unspecified() == false
            {
                tcp_socket
                    .bind(*self.tap_data.get_bind_addr())
                    .map_err(|e|
                        map_error!("cannot bind tcp socket '{}', error: {}", self.tap_data.get_remote_addr(), e)
                    )?;
            }
                
            let socket = 
                if let Some(c_timeout) = self.tap_data.get_conn_timeout()
                {
                    time::timeout(c_timeout, tcp_socket.connect(*self.tap_data.get_remote_addr()))
                        .await
                        .map_err(|e|
                            map_error!("can not connect to tcp '{}', error: {}", self.tap_data.get_remote_addr(), e)
                        )?
                }
                else
                {
                    tcp_socket
                        .connect(*self.tap_data.get_remote_addr())
                        .await
                }
                .map_err(|e|
                    map_error_os!(e, "can not connect to tcp '{}', error: {}", self.tap_data.get_remote_addr(), e)
                )?;


            // set socket
            self.sock = Some(socket);

            return Ok(());
        }

        #[inline]
        async 
        fn send(&mut self, msg: &[u8]) -> std::io::Result<usize> 
        {
            let sock = 
                self.sock
                    .as_mut()
                    .ok_or_else(||
                        std::io::Error::new(ErrorKind::NotConnected, "no connection")
                    )?;

            sock.write_all(msg).await?;
            sock.flush().await?;

            return Ok(msg.len());
        }

        async 
        fn disconnectlog(&mut self) -> std::io::Result<()> 
        {
            match self.sock.take()
            {
                Some(mut s) => 
                {
                    s.shutdown().await?;

                    drop(s);

                    Ok(())
                },
                None =>
                {
                    Ok(())
                }
            }
        }

        #[inline]
        fn is_connected(&self) -> bool 
        {
            return self.sock.is_some();
        }

        #[inline]
        fn get_type(&self) -> TapType 
        {
            return self.cur_tap_type;
        }

        #[inline]
        fn get_max_msg_size() -> usize
        {
            return crate::SyslogNetTcp::get_max_msg_len();
        }

        #[inline]
        fn update_tap_data(&mut self, tap_data: SyslogNetTcp)
        {
            self.tap_data = tap_data;
        }
    }

    impl AsyncSyslogTap<SyslogNetUdp> for AsyncTap<UdpSocket, SyslogNetUdp>
    {
        fn new(req_tap: SyslogNetUdp) -> SyRes<Self>
        {
            let ret = 
                Self
                {
                    sock: None, 
                    tap_data: req_tap,
                    cur_tap_type: TapType::NetUdp,
                };

            return Ok(ret);
        }

        async 
        fn connectlog(&mut self) -> SyRes<()> 
        {
            let socket = 
                UdpSocket
                    ::bind(self.tap_data.get_bind_addr())
                        .await
                        .map_err(|e|
                            map_error_os!(e, "can not bind udp '{}', error: {}", self.tap_data.get_bind_addr(), e)
                        )?;

            socket
                .connect(self.tap_data.get_remote_addr())
                .await
                .map_err(|e|
                    map_error_os!(e, "can not connect to udp '{}', error: {}", self.tap_data.get_remote_addr(), e)
                )?;

            // set socket
            self.sock = Some(socket);

            return Ok(());
        }

        #[inline]
        async 
        fn send(&mut self, msg: &[u8]) -> std::io::Result<usize> 
        {
            let sock = 
                self
                    .sock
                    .as_mut()
                    .ok_or_else(||
                        std::io::Error::new(ErrorKind::NotConnected, "no connection")
                    )?;

            return sock.send(msg).await;
        }

        async
        fn disconnectlog(&mut self) -> std::io::Result<()> 
        {
            match self.sock.take()
            {
                Some(s) => 
                {
                    drop(s);

                    Ok(())
                },
                None =>
                {
                    Ok(())
                }
            }
        }

        #[inline]
        fn is_connected(&self) -> bool 
        {
            return self.sock.is_some();
        }

        #[inline]
        fn get_type(&self) -> TapType 
        {
            return self.cur_tap_type;
        }

        #[inline]
        fn get_max_msg_size() -> usize
        {
            return crate::SyslogNetUdp::get_max_msg_len();
        }

        #[inline]
        fn update_tap_data(&mut self, tap_data: SyslogNetUdp)
        {
            self.tap_data = tap_data;
        }
    }
}


impl AsyncSyslogTap<SyslogLocal> for AsyncTap<UnixDatagram, SyslogLocal>
{

    fn new(req_tap: SyslogLocal) -> SyRes<Self>
    {
        return Ok(
            Self
            {
                sock: None, 
                tap_data: req_tap,
                cur_tap_type: TapType::None,
            }
        );
    }


    async 
    fn connectlog(&mut self) -> SyRes<()> 
    {
        let sock = 
            UnixDatagram
                ::unbound()
                    .map_err(|e|
                        map_error_os!(e, "unbounded unix datagram initialization failure: {}", e)
                    )?;

        let tap_type = 
            if self.tap_data.get_use_alternative() == false && self.tap_data.get_custom_remote_path().is_some() == true 
            {
                if let Err(e) = sock.connect(self.tap_data.get_custom_remote_path().as_ref().unwrap())
                {
                    throw_error_os!(e, "failed to open connection to syslog server at '{}'", 
                        self.tap_data.get_custom_remote_path().as_ref().unwrap().display());
                }
                else 
                {
                    TapType::CustomLog
                }
            }
            else if self.tap_data.get_custom_remote_path().is_some() == true &&
                sock.connect(self.tap_data.get_custom_remote_path().as_ref().unwrap()).is_ok() == true
            {
                TapType::CustomLog
            }
            else if let Ok(_) = sock.connect(PATH_LOG_PRIV)
            {
                TapType::Priv
            }
            else if let Ok(_) = sock.connect(PATH_LOG)
            {
                TapType::UnPriv
            }
            else if let Ok(_) = sock.connect(PATH_OLDLOG)
            {
                TapType::OldLog
            }
            else if let Ok(_) = sock.connect(PATH_OSX)
            {
                TapType::Priv
            }
            else
            {
                // failed to open socket
                throw_error_errno!(Errno::last(), "failed to open connection to syslog server");
            };

        #[cfg(any(
            target_os = "freebsd",
            target_os = "dragonfly",
            target_os = "openbsd",
            target_os = "netbsd",
            target_os = "macos"
        ))]
        {
            use std::os::fd::AsRawFd;

            // setting sendbuf
            let mut len: std::mem::MaybeUninit<nix::libc::socklen_t> = std::mem::MaybeUninit::uninit();

            // set the sndbuf len
            let res = 
                unsafe
                {
                    nix::libc::getsockopt(
                        sock.as_raw_fd(), 
                        nix::libc::SOL_SOCKET, 
                        nix::libc::SO_SNDBUF, 
                        len.as_mut_ptr() as *mut nix::libc::c_void, 
                        &mut { 
                            std::mem::size_of::<nix::libc::socklen_t>() as nix::libc::socklen_t
                        } 
                    )
                };
    
            if res == 0
            {
                let mut len = unsafe { len.assume_init() } as usize;

                if len < MAXLINE
                {
                    len = MAXLINE;

                    unsafe {
                        nix::libc::setsockopt(
                            sock.as_raw_fd(), 
                            nix::libc::SOL_SOCKET, 
                            nix::libc::SO_SNDBUF, 
                            &len as *const _ as *const nix::libc::c_void, 
                            std::mem::size_of::<nix::libc::socklen_t>() as nix::libc::socklen_t
                        )
                    };
                }
            }
        }

        sock
            .shutdown(std::net::Shutdown::Read)
            .map_err(|e|
                map_error_os!(e, "can not shutdown read portion, error: '{}'", e)
            )?;

        self.sock = Some(sock);
        self.cur_tap_type = tap_type;

        return Ok(());
    }

    #[inline]
    async 
    fn send(&mut self, msg: &[u8]) -> std::io::Result<usize> 
    {
        let sock = 
            self.sock
                .as_mut()
                .ok_or_else(||
                    std::io::Error::new(ErrorKind::NotConnected, "no connection")
                )?;

        return sock.send(msg).await;
    }

    async 
    fn disconnectlog(&mut self) -> std::io::Result<()>
    {
        match self.sock.take()
        {
            Some(s) => 
            {
                self.cur_tap_type = TapType::None;

                s.shutdown(Shutdown::Both)
            },
            None =>
            {
                self.cur_tap_type = TapType::None;

                Ok(())
            }
        }
    }

    #[inline]
    fn is_connected(&self) -> bool
    {
        return self.sock.is_some();
    }

    #[inline]
    fn get_type(&self) -> TapType 
    {
        return self.cur_tap_type;
    }

    #[inline]
    fn get_max_msg_size() -> usize
    {
        return crate::SyslogLocal::get_max_msg_len();
    }

    #[inline]
    fn update_tap_data(&mut self, tap_data: SyslogLocal)
    {
        self.tap_data = tap_data;
    }
}



#[cfg(feature = "build_ext_file")]
mod with_file
{
    use std::io::ErrorKind;

    use tokio::fs::File;

    use crate::{a_sync::{AsyncSyslogTap, AsyncTap}, error::SyRes,  map_error_os, SyslogDestMsg, SyslogFile, TapType};

    impl AsyncSyslogTap<SyslogFile> for AsyncTap<File, SyslogFile>
    {
        fn new(req_tap: SyslogFile) -> SyRes<Self>
        {
            let ret = 
                Self
                {
                    sock: None, 
                    tap_data: req_tap,
                    cur_tap_type: TapType::LocalFile,
                };

            return Ok(ret);
        }

        async 
        fn connectlog(&mut self) -> SyRes<()> 
        {

            let file = 
                File::options()
                    .append(true)
                    .read(true)
                    .write(true)
                    .create(true)
                    .open(self.tap_data.get_path())
                    .await
                    .map_err(|e|
                        map_error_os!(e, "can not open file '{}' to write, error: '{}'", 
                            self.tap_data.get_path().display(), e)
                    )?;

            self.sock = Some(file);

            return Ok(());
        }

        #[inline]
        async 
        fn send(&mut self, msg: &[u8]) -> std::io::Result<usize> 
        {
            use crate::tokio::io::AsyncWriteExt;
            
            let sock = 
                self
                    .sock
                    .as_mut()
                    .ok_or_else(||
                        std::io::Error::new(ErrorKind::NotConnected, "no connection")
                    )?;
            
            sock.write_all(msg).await?;

            return Ok(msg.len());
        }

        async 
        fn disconnectlog(&mut self) -> std::io::Result<()> 
        {
            match self.sock.take()
            {
                Some(s) => 
                {
                    self.cur_tap_type = TapType::None;

                    s.sync_data().await?;

                    drop(s);

                    return Ok(());
                },
                None =>
                {
                    self.cur_tap_type = TapType::None;

                    return Ok(())
                }
            }
        }

        #[inline]
        fn is_connected(&self) -> bool 
        {
            return self.sock.is_some();
        }

        #[inline]
        fn get_type(&self) -> TapType 
        {
            return self.cur_tap_type;
        }

        #[inline]
        fn get_max_msg_size() -> usize
        {
            return crate::SyslogFile::get_max_msg_len();
        }

        #[inline]
        fn update_tap_data(&mut self, tap_data: SyslogFile)
        {
            self.tap_data = tap_data;
        }
    }
}