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. The MIT License (MIT)
 *                     
 *   3. EUROPEAN UNION PUBLIC LICENCE v. 1.2 EUPL © the European Union 2007, 2016
 */




#[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;
        }
    }
}


#[cfg(target_family = "unix")]
mod with_unix_syslog
{
    use std::io::ErrorKind;
    use std::net::Shutdown;

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

    use crate::a_sync::{AsyncSyslogTap, AsyncTap};
    use crate::{SyslogDestMsg, SyslogLocal, TapType, map_error_os, throw_error_errno, throw_error_os};
    use crate::common::*;
    use crate::error::SyRes;


    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;
        }
    }
}


// --- windows ---

#[cfg(target_family = "windows")]
mod with_windows_eventlog
{
    use std::{fmt};
    use std::io::ErrorKind;

    use crate::portable::{EventLogLocal};
    use crate::{error::SyRes, TapType};


    use crate::
    {
        AsyncSyslogTap, AsyncTap, SyslogDestMsg, WindowsEvent, map_error_os
    };

    use crate::common::*;

    impl AsyncSyslogTap<WindowsEvent> for AsyncTap<EventLogLocal, WindowsEvent>
    {
        /// Creates the instance for local socket.
        /// 
        /// If `opt_path` does not exist or can not be opened, in case if 
        /// flag `use_altern` is set to true, other known localions will 
        /// be attempted to connect before giving up.
        /// 
        /// # Arguments
        /// 
        /// * `req_tap` - a type of the tap which is required. See [TapTypeData] 
        ///     The actual type of the tap selected will be stored in the field 
        ///     `cur_tap_type`.
        fn new(req_tap: WindowsEvent) -> SyRes<Self>
        {
            return Ok(
                Self
                {
                    sock: None, 
                    tap_data: req_tap,
                    cur_tap_type: TapType::None,
                }
            );
        }

        async 
        fn connectlog(&mut self) -> SyRes<()> 
        {
            let sock = 
                EventLogLocal::new(&self.tap_data)
                    .map_err(|e|
                        map_error_os!(e, "EventRegister error: '{}'", e)
                    )?;

            self.sock = Some(sock);
            self.cur_tap_type = TapType::WindowsEventLog;

            return Ok(());
        }

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

            return sock.send(msg);
        }

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

                    drop(s);

                    Ok(())
                },
                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 WindowsEvent::get_max_msg_len();
        }

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