syslog_rs/sync/

syslog_sync_internal.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
 */


use std::io::Error;

use instance_copy_on_write::{ICoW, ICoWRead};

use crate::formatters::{SyslogFormatted, SyslogFormatter};
use crate::portable;
use crate::common::*;
use crate::error::{SyRes, SyslogError};
use crate::SyslogDestination;

use super::socket::*;

#[cfg(target_family = "unix")]
use nix::libc;

#[derive(Debug, Clone)]
pub(crate) struct LogItems
{
    /// An identification i.e program name, thread name
    pub(crate) logtag: String, 

    /// A pid of the program.
    pub(crate) logpid: String,

    /// A logmask
    pub(crate) logmask: i32,

    pub(crate) logstat: LogStat,

    /// Holds the facility 
    pub(crate) facility: LogFacility,
}

impl LogItems
{
    #[inline]
    pub 
    fn new(ident: Option<&str>, logmask: i32, logstat: LogStat, facility: LogFacility) -> Self
    {
        let mut log_inst = 
            Self
            {
                logtag: String::new(),
                logpid: portable::get_pid().to_string(),
                logmask: logmask,
                logstat: logstat,
                facility: LogFacility::empty(),
            };

        log_inst.set_log_facility(facility);
        log_inst.set_identity(ident);

        return log_inst;
    }

    pub(crate) 
    fn set_log_facility(&mut self, facility: LogFacility)
    {
        let log_facility =
            if facility.is_empty() == false && 
                (facility & !LogMask::LOG_FACMASK).is_empty() == true
            {
                facility
            }
            else
            {
                // default facility code
                LogFacility::LOG_USER
            };

        self.facility = log_facility;

        return;
    }

    pub(crate) 
    fn set_identity(&mut self, ident: Option<&str>)
    {
        let logtag = 
            match ident
            {
                Some(r) => 
                    truncate_n(r, RFC_MAX_APP_NAME).to_string(),
                None => 
                    truncate_n(
                        portable::p_getprogname()
                            .unwrap_or("notavail".to_string())
                            .as_str(),
                        RFC_MAX_APP_NAME
                    )
                    .to_string()
            };

        self.logtag = logtag;

        return;
    }

    #[inline]
    pub 
    fn get_progname(&self) -> &str
    {
        return &self.logtag;
    }

    #[inline]
    pub 
    fn get_pid(&self) -> &str
    {
        return &self.logtag;
    }

    #[inline]
    fn is_logmasked(&self, pri: Priority) -> bool
    {
        return ((1 << (pri & LogMask::LOG_PRIMASK)) & self.logmask) == 0;
    }

    pub(crate) 
    fn set_logmask(&mut self, logmask: i32) -> i32
    {
        let oldmask = self.logmask;

        if logmask != 0
        {
            self.logmask = logmask;
        }

        return oldmask;
    }

    pub(crate) 
    fn vsyslog1_msg<F, D>(&self, mut pri: Priority, fmt: &F) ->  Option<(SyslogFormatted, LogStat)>
    where F: SyslogFormatter, D: SyslogDestination
    {
        // check for invalid bits
        if let Err(e) = pri.check_invalid_bits()
        {
            self.logstat.send_to_stderr(&e.to_string());
        }

        // check priority against setlogmask
        if self.is_logmasked(pri) == true
        {
            return None;
        }

        // set default facility if not specified in pri
        if (pri.bits() & LOG_FACMASK) == 0
        {
            pri.set_facility(self.facility);
        }

        // set PID if needed
        let msg_pid = 
            if self.logstat.intersects(LogStat::LOG_PID) == true
            {
                Some(self.logpid.as_str())
            }
            else
            {
                None
            };

        // format msg
        let msg_formatted = 
            fmt.vsyslog1_format(D::SocketTap::get_max_msg_size(), pri, &self.logtag, msg_pid);
        
        // output to stderr if required
        self.logstat.send_to_stderr(msg_formatted.get_stderr_output());

        return Some((msg_formatted, self.logstat));
    }
}

#[derive(Debug)]
pub(crate) struct SyslogSocket<D: SyslogDestination>
{
    stream: ICoW<D::SocketTap>
}

impl<D: SyslogDestination> SyslogSocket<D>
{
    pub(crate) 
    fn new(logstat: LogStat, net_tap_prov: D) -> SyRes<Self>
    {
        let mut sock = 
            D::SocketTap::new(net_tap_prov)?;

        if logstat.contains(LogStat::LOG_NDELAY) == true
        {
            sock.connectlog()?;
        }

        return Ok(
            Self{ stream: ICoW::new(sock) }
        );
    }

    #[inline]
    pub(crate) 
    fn update_tap_data(&self, tap_data: D) -> SyRes<()>
    {
        
        let mut lock = 
            self
                .stream
                .clone_copy_exclusivly();

        //lock.disconnectlog()?;

        lock.update_tap_data(tap_data);

        lock.connectlog()?;

        lock.commit();
        
        return Ok(());
    }

    #[inline]
    pub(crate) 
    fn reconnectlog(&self) -> SyRes<()>
    {
        let mut lock = 
            self
                .stream
                .clone_copy_exclusivly();

        //lock.disconnectlog()?;
        lock.connectlog()?;

        lock.commit();

        return Ok(());
    }

    #[inline]
    pub(crate) 
    fn connectlog(&self) -> SyRes<()>
    {
        let mut lock =  
            self
                .stream
                .clone_copy_exclusivly();

        lock.connectlog()?;

        lock.commit();

        return Ok(());
    }

    /// Disconnects the unix stream from syslog.
    /// Do not call this function from internal.
    #[inline]
    pub(crate) 
    fn disconnectlog(&self) -> SyRes<()>
    {
        let lock =  
            self
                .stream
                .clone_copy_exclusivly();

        lock.commit();

        return Ok(());
    }

    /// A `send` error handler which is specific for `unix` OSes.
    #[cfg(target_family = "unix")]
    fn handle_error<'h>(&'h self, stream: ICoWRead<'h, <D as SyslogDestination>::SocketTap>, 
        e: Error) -> Result<ICoWRead<'h, D::SocketTap>, SyslogError>
    {
        use std::io::ErrorKind;
        use std::thread::sleep;
        use std::time::Duration;

        // The socket is not connected.
        if e.kind() == ErrorKind::NotConnected 
        {
            match self.stream.clone_copy_exclus_or_read()
            {
                Ok(mut lock) =>
                {
                    lock.connectlog()?;

                    lock.commit();

                    return Ok(self.stream.read());
                },
                Err(read) =>
                {
                    return Ok(read);
                }
            }
        }
        else 
        {
            if D::DEST_TYPE.is_network() == false
            {
                if let Some(libc::ENOBUFS) = e.raw_os_error()
                {
                    // scenario 2, not enoughspace
                    if stream.get_type().is_priv() == true
                    {
                        use nix::errno::Errno;

                        use crate::throw_error_errno;

                        throw_error_errno!(Errno::ENOBUFS, "Not enough space in priv socket.");
                    }

                    sleep(Duration::from_micros(1));

                    return Ok(stream);
                }
                else
                {
                    // scenario 1
                    match self.stream.clone_copy_exclus_or_read()
                    {
                        Ok(mut lock) =>
                        {
                            lock.connectlog()?;

                            lock.commit();

                            return Ok(self.stream.read());
                        },
                        Err(read) =>
                        {
                            return Ok(read);
                        }
                    }

                    // if resend will fail then probably the scn 2 will take place
                }   
            }
            else
            {
                match self.stream.clone_copy_exclus_or_read()
                {
                    Ok(mut lock) =>
                    {
                        lock.connectlog()?;

                        lock.commit();

                        return Ok(self.stream.read());
                    },
                    Err(read) =>
                    {
                        return Ok(read);
                    }
                }
            }
        }
    }

    /// A `send` error handler which is specific for `windows` OS.
    #[cfg(target_family = "windows")]
    fn handle_error<'h>(&'h self, _stream: ICoWRead<'h, <D as SyslogDestination>::SocketTap>, 
        e: Error) -> Result<ICoWRead<'h, D::SocketTap>, SyslogError>
    {
        use crate::DestinationType;
        use crate::error::SyslogErrCode;

        if D::DEST_TYPE == DestinationType::Local
        {
            // unwrap sshould never panic because `from_raw_os_error` is used.
            let Ok(werr) = e.downcast::<windows::core::Error>()
                else
                {
                    

                    return Err(
                        SyslogError::new(SyslogErrCode::InternalError, 
                            "error downcast failed".into())
                    );
                };

            return Err(
                SyslogError::new(
                    SyslogErrCode::InternalError, 
                    werr.message()
                )
            );
        }
        else if D::DEST_TYPE == DestinationType::Network
        {
            match self.stream.clone_copy_exclus_or_read()
            {
                Ok(mut lock) =>
                {
                    lock.connectlog()?;

                    lock.commit();

                    return Ok(self.stream.read());
                },
                Err(read) =>
                {
                    return Ok(read);
                }
            }
        }
        else
        {
            return Err(
                SyslogError::new(
                    SyslogErrCode::InternalError, 
                    format!("IO error: {}", e)
                )
            );
        }

    }


    /// There are two possible scenarios when send may fail:
    /// 1. syslog temporary unavailable
    /// 2. syslog out of buffer space
    /// If we are connected to priv socket then in case of 1 we reopen connection
    ///      and retry once.
    /// If we are connected to unpriv then in case of 2 repeatedly retrying to send
    ///      until syslog socket buffer space will be cleared
    /// What about the rest scenarios???
    pub(crate) 
    fn vsyslog1(&self, logstat: LogStat, mut msg_formatted: SyslogFormatted)
    {
        let fullmsg = msg_formatted.get_full_msg();

        let mut stream = self.stream.read();

        loop
        {
            match stream.send(fullmsg.as_bytes())
            {
                Ok(_) => 
                    return,
                Err(e) =>
                {
                    match self.handle_error(stream, e)
                    {
                        Ok(read) => 
                        {
                            stream = read;

                            continue;
                        },
                        Err(e) =>
                        {
                            logstat.send_to_stderr(&format!("connectlog() failed with {}", e));
                            break;
                        }
                    }
                }
            }
        }
        

        // If program reached this point then transmission over socket failed.
        // Try to output message to console

        let _ = logstat.send_to_syscons(msg_formatted.get_stderr_output());

        return;
    }

}

#[cfg(any(feature = "build_with_thread_local", feature = "build_with_queue"))]
pub mod lockless
{
    use std::io::ErrorKind;
    use std::thread::sleep;
    use std::time::Duration;

    use super::*;
    
    #[derive(Debug)] 
    pub(crate) struct SyslogSocketLockless<D: SyslogDestination>
    {
        stream: D::SocketTap
    }

    impl<D: SyslogDestination> SyslogSocketLockless<D>
    {
        pub(crate) 
        fn new(logstat: LogStat, net_tap_prov: D) -> SyRes<Self>
        {
            let mut sock = 
                D::SocketTap::new(net_tap_prov)?;

            if logstat.contains(LogStat::LOG_NDELAY) == true
            {
                sock.connectlog()?;
            }

            return Ok(
                Self{ stream: sock }
            );
        }

        #[inline]
        pub(crate) 
        fn update_tap_data(&mut self, tap_data: D) -> SyRes<()>
        {
            self.stream.disconnectlog()?;
            self.stream.update_tap_data(tap_data);

            return self.stream.connectlog();
        }

        #[inline]
        pub(crate) 
        fn reconnectlog(&mut self) -> SyRes<()>
        {
            self.stream.disconnectlog()?;
            self.stream.connectlog()?;

            return Ok(());
        }

        #[inline]
        pub(crate) 
        fn connectlog(&mut self) -> SyRes<()>
        {
            return self.stream.connectlog();
        }

            /// Disconnects the unix stream from syslog.
        /// Do not call this function from internal.
        #[inline]
        pub(crate) 
        fn disconnectlog(&mut self) -> SyRes<()>
        {
            return self.stream.disconnectlog();
        }

        

        /// There are two possible scenarios when send may fail:
        /// 1. syslog temporary unavailable
        /// 2. syslog out of buffer space
        /// If we are connected to priv socket then in case of 1 we reopen connection
        ///      and retry once.
        /// If we are connected to unpriv then in case of 2 repeatedly retrying to send
        ///      until syslog socket buffer space will be cleared
        /// What about the rest scenarios??? 
        pub(crate) 
        fn vsyslog1(&mut self, logstat: LogStat, mut msg_formatted: SyslogFormatted)
        {
            if self.stream.is_connected() == false
            {
                if let Err(e) = self.stream.connectlog()
                {
                    logstat.send_to_stderr(&e.into_inner());
                }
            }

            let fullmsg = msg_formatted.get_full_msg();

            let mut repeated = false;
            loop
            {
                match self.stream.send(fullmsg.as_bytes())
                {
                    Ok(_) => 
                        return,
                    Err(ref e) if e.kind() == ErrorKind::NotConnected => 
                    {
                        break;
                    },
                    // poisoned mutex
                    Err(ref e) if e.kind() == ErrorKind::Deadlock =>
                    {
                        logstat.send_to_stderr(&e.to_string());
                        break;
                    },
                    Err(err) =>
                    {
                        if D::DEST_TYPE.is_network() == false
                        {
                            #[cfg(target_family = "unix")]
                            {
                                if let Some(libc::ENOBUFS) = err.raw_os_error()
                                {
                                    // scenario 2, not enoughspace
                                    if self.stream.get_type().is_priv() == true
                                    {
                                        break;
                                    }

                                    sleep(Duration::from_micros(1));
                                }
                                else
                                {
                                    // scenario 1
                                    let _ = self.stream.disconnectlog();
                                    if let Err(e) = self.stream.connectlog()
                                    {
                                        logstat.send_to_stderr(&e.into_inner());
                                        break;
                                    }
                                }
                            }

                            #[cfg(target_family = "windows")]
                            {
                                let Ok(werr) = err.downcast::<windows::core::Error>()
                                    else
                                    {
                                        logstat.send_to_stderr("error downcast failed");
                                        break;
                                    };
                                    
                                logstat.send_to_stderr(&werr.message());
                                break;
                            }
            
                            // if resend will fail then probably the scn 2 will take place  
                        }
                        else
                        {
                            if repeated == false
                            {
                                repeated = true;

                                let _ = self.stream.disconnectlog();
                                if let Err(e) = self.stream.connectlog()
                                {
                                    logstat.send_to_stderr(&e.into_inner());
                                    break;
                                }
                            }
                            else
                            {
                                logstat.send_to_stderr("syslog: can not send to remote server");
                                break;
                            }
                        }
                        
                    }
                }
            }
            

            // If program reached this point then transmission over socket failed.
            // Try to output message to console

            let _ = logstat.send_to_syscons(msg_formatted.get_stderr_output());

            return;
        }

    }
}

#[cfg(any(feature = "build_with_thread_local", feature = "build_with_queue"))]
pub use self::lockless::*;



#[cfg(test)]
mod tests
{
    use crate::LOG_MASK;

    use super::*;

    #[test]
    fn test_log_cons()
    {

        
        //let net_tap = TapTypeData::new_unix(None, true).unwrap();
        
        let msg = "header msg message payload";

        let lsts = LogStat::LOG_CONS;
        
        lsts.send_to_syscons(msg);

        return;
    }

    #[test]
    fn test_bit_operations()
    {

        let correct = 
            LogItems::new(Some("test1"), 0xFF, LogStat::LOG_PID, LogFacility::LOG_DAEMON);

        assert_eq!(correct.facility, LogFacility::LOG_DAEMON);
        assert_eq!((correct.facility & !LogFacility::LOG_DAEMON), LogFacility::empty());
    }

    #[test]
    fn test_bit_operations2()
    {

        let mut pri = Priority::LOG_ALERT;

        let res = pri.check_invalid_bits();

        assert_eq!(res.is_ok(), true);
        assert_eq!(pri.bits(), Priority::LOG_ALERT.bits());
    }

    #[test]
    fn test_set_priority()
    {
        let mut correct = 
            LogItems::new(Some("test1"), 0xFF, LogStat::LOG_PID, LogFacility::LOG_DAEMON);

        let ret = correct.set_logmask(LOG_MASK!(Priority::LOG_ERR));

        assert_eq!(ret, 0xff);

        let ret = correct.set_logmask(LOG_MASK!(Priority::LOG_ERR));

        assert_eq!(ret, LOG_MASK!(Priority::LOG_ERR));

        let ret = correct.is_logmasked(Priority::LOG_WARNING);
        assert_eq!(ret, true);

        let ret = correct.is_logmasked(Priority::LOG_ERR);
        assert_eq!(ret, false);

        let ret = correct.is_logmasked(Priority::LOG_CRIT);
        assert_eq!(ret, true);
    }
}