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


use std::io::ErrorKind;
use std::thread::sleep;
use std::time::Duration;


use instance_copy_on_write::ICoW;
use nix::libc;


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

use super::socket::*;


#[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: i32) -> 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.bits()) == true
        {
            return None;
        }

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

        // format msg
        let msg_formatted = 
            fmt.vsyslog1_format(D::SocketTap::get_max_msg_size(), pri, &self.logtag, &self.logpid);
        
        // 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
                .try_clone_exclusivly()
                .ok_or_else(||
                    map_error_code!(CoWAlreadyLocked, "cannot exclusivly lock instance, already locked")
                )?;

        //lock.disconnectlog()?;

        lock.update_tap_data(tap_data);

        lock.connectlog()?;

        return 
            lock
                .commit()
                .map_err(|_|
                    map_error_code!(CoWWriteError, "updating tap data failed!")
                )
    }

    #[inline]
    pub(crate) 
    fn reconnectlog(&self) -> SyRes<()>
    {
        let mut lock = 
            self
                .stream
                .try_clone_exclusivly()
                .ok_or_else(||
                    map_error_code!(CoWAlreadyLocked, "cannot exclusivly lock instance, already locked")
                )?;

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

        return 
            lock
                .commit()
                .map_err(|_|
                    map_error_code!(CoWWriteError, "reconnect failed!")
                );
    }

    #[inline]
    pub(crate) 
    fn connectlog(&self) -> SyRes<()>
    {
        let mut lock =  
            self
                .stream
                .try_clone_exclusivly()
                .ok_or_else(||
                    map_error_code!(CoWAlreadyLocked, "cannot exclusivly lock instance, already locked")
                )?;
                /*.map_err(|e|
                    map_error_code!(CoWAlreadyLocked, "cannot exclusivly lock instance, err: {}", e)
                )?
                .connectlog(); */

        lock.connectlog()?;

        return 
            lock
                .commit()
                .map_err(|_|
                    map_error_code!(CoWWriteError, "reconnect failed!")
                );
    }

        /// Disconnects the unix stream from syslog.
    /// Do not call this function from internal.
    #[inline]
    pub(crate) 
    fn disconnectlog(&self) -> SyRes<()>
    {
        let lock =  
            self
                .stream
                .try_clone_exclusivly()
                .ok_or_else(||
                    map_error_code!(CoWAlreadyLocked, "cannot exclusivly lock instance, already locked")
                )?;

                /*.map_err(|e|
                    map_error_code!(CoWAlreadyLocked, "cannot exclusivly lock instance, err: {}", e)
                )?
                .disconnectlog();*/

        return 
            lock
                .commit()
                .map_err(|_|
                    map_error_code!(CoWWriteError, "reconnect failed!")
                );
    }

    /// 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 repeated = false;
        loop
        {
            let (tap_type, res) = 
                {
                    let read0 = self.stream.read();

                    (read0.get_type(), read0.send(fullmsg.as_bytes()))
                };

            match res
            {
                Ok(_) => 
                    break,
                Err(ref e) if e.kind() == ErrorKind::NotConnected => 
                {
                    if let Err(e) = self.connectlog()
                    {
                        if e.get_errcode() == SyslogErrCode::CoWAlreadyLocked
                        {
                            continue;
                        }
                        else
                        {
                            logstat.send_to_stderr(&format!("connectlog() failed with {}", e));
                            break;
                        }
                    }
                },
                Err(err) =>
                {
                    if D::DEST_TYPE.is_network() == false
                    {
                        if let Some(libc::ENOBUFS) = err.raw_os_error()
                        {
                            // scenario 2, not enoughspace
                            if tap_type.is_priv() == true
                            {
                                break;
                            }

                            sleep(Duration::from_micros(1));
                        }
                        else
                        {
                            // scenario 1
                            if let Err(e) = self.reconnectlog()
                            {
                                if e.get_errcode() == SyslogErrCode::CoWAlreadyLocked
                                {
                                    continue;
                                }
                                else
                                {
                                    logstat.send_to_stderr(&format!("reconnectlog() failed with {}", e));
                                    break;
                                }
                            }
        
                            // if resend will fail then probably the scn 2 will take place
                        }   
                    }
                    else
                    {
                        if let Err(e) = self.connectlog()
                        {
                            if e.get_errcode() == SyslogErrCode::CoWAlreadyLocked
                            {
                                continue;
                            }
                            else
                            {
                                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 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(_) => 
                        break,
                    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());
                        return;
                    },
                    Err(err) =>
                    {
                        if D::DEST_TYPE.is_network() == false
                        {
                            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;
                                }
                            }
            
                                // 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.bits());
        assert_eq!(ret, true);

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

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