syslog_rs/a_sync/

syslog_async_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::fmt;
use std::marker::PhantomData;

use crate::a_sync::syslog_trait::AsyncSyslogApi;
use crate::formatters::SyslogFormatter;
use crate::map_error_os;
use crate::portable;
use crate::common::*;
use crate::error::SyRes;
use crate::socket::TapType;
use crate::AsyncSyslogDestination;

use super::async_socket::*;

/// A trait which generalize some operations which are different from the
/// various async providers i.e `smol` or `tokio` or external.
#[allow(async_fn_in_trait)]
pub trait AsyncSyslogInternalIO: fmt::Debug + Send + 'static
{
    /// Sends a message to syscons device. Usually this is a `/dev/console`
    /// defined by crate::common::PATH_CONSOLE.
    /// 
    /// # Arguemnts
    /// 
    /// * `logstat` - a instance setup [LogStat].
    /// 
    /// * `msg_payload` - a payload of the syslog message (without headers).
    async fn send_to_syscons(logstat: LogStat, msg_payload: &str);

    /// Sends a message to stderr device.
    /// 
    /// # Arguemnts
    /// 
    /// * `logstat` - a instance setup [LogStat].
    /// 
    /// * `msg` - a payload of the syslog message (without headers).
    async fn send_to_stderr(logstat: LogStat, msg: &str);

    /// Sleep the current task for `us` microseconds.
    /// 
    /// # Arguments
    /// 
    /// * `us` - microseconds.
    async fn sleep_micro(us: u64);
}

/// A trait which generalize the mutex from the std lib's of multiple async executors.
/// The trait should be implemented on the mutex direclty.
#[allow(async_fn_in_trait)]
pub trait AsyncMutex<F: SyslogFormatter, D: AsyncSyslogDestination, DS: AsyncSyslogApi<F, D>>
{
    /// A mutex guard type.
    type MutxGuard<'mux>: AsyncMutexGuard<'mux, F, D, DS> where Self: 'mux;

    /// Creates new mutex instance for type which implements the [AsyncSyslogApi].
    fn a_new(v: DS) -> Self;

    /// Locks the mutex emmiting the `mutex guard`.
    async fn a_lock<'mux>(&'mux self) -> Self::MutxGuard<'mux>;
}

/// A trait which generalize the mutex guarding emited by the mutex from various async executors.
pub trait AsyncMutexGuard<'mux, F: SyslogFormatter, D: AsyncSyslogDestination, DS: AsyncSyslogApi<F, D>>
{
    /// Returns the reference to the inner type of the mutex guard.
    fn guard(&self) -> &DS;

    /// Returns the mutable reference to the inner type of the mutex guard.
    fn guard_mut(&mut self) -> &mut DS;
}

/// Internal structure of the syslog async client.
#[derive(Debug)]
pub struct AsyncSyslogInternal<F: SyslogFormatter + Send, D: AsyncSyslogDestination, IO: AsyncSyslogInternalIO>
{
    /// An identification i.e program name, thread name
    logtag: String, 

    /// A pid of the program.
    logpid: String,

    /// Defines how syslog operates
    logstat: LogStat,

    /// Holds the facility 
    facility: LogFacility,

    /// A logmask
    logmask: i32,

    /// A stream
    stream: D::SocketTap,

    /// Phantom for [SyslogFormatter]
    _p: PhantomData<F>,

    /// Phantom for the [AsyncSyslogInternalIO] which provides writing to console and other IO.
    _p2: PhantomData<IO>,
}


impl<F: SyslogFormatter + Send, D: AsyncSyslogDestination, IO: AsyncSyslogInternalIO> AsyncSyslogInternal<F, D, IO>
{
    /// Creates new instance of [SyslogInternal] which contains all 
    /// client syslog logic.
    ///
    /// # Arguments
    ///
    /// * `ident` - An optional argument which takes ref to str. If none, the
    ///     ident will be set later. Yje ident will be trucated to 48 UTF8
    ///     chars.
    /// 
    /// * `logstat` - A [LogStat] flags separated by '|'
    /// 
    /// * `facility` - A [LogFacility] flag
    /// 
    /// * `req_tap` - A type of the syslog instance reuired. See [SyslogDestination].
    /// 
    /// # Returns
    ///
    /// A [SyRes] is returned with following:
    /// 
    /// * [Result::Ok] - with the instance.
    /// 
    /// * [Result::Err] - with error description.
    pub(crate) 
    fn new(
        ident: Option<&str>, 
        logstat: LogStat, 
        facility: LogFacility, 
        req_tap: D
    ) -> SyRes<Self>
    {
        // check if log_facility is invalid
        let log_facility =
            if facility.is_empty() == false && 
                (facility & !LogMask::LOG_FACMASK).is_empty() == true
            {
                facility
            }
            else
            {
                // default facility code
                LogFacility::LOG_USER
            };

        let logtag = 
            match ident
            {
                Some(r) => 
                    truncate_n(r, RFC_MAX_APP_NAME).to_string(),
                None => 
                    truncate_n(
                        portable::p_getprogname()
                            .unwrap_or("".to_string())
                            .as_str(),
                        RFC_MAX_APP_NAME
                    )
                    .to_string()
            };

        let sock = D::SocketTap::new(req_tap)?;

        return Ok(
            Self
            {
                logtag: logtag,
                logpid: portable::get_pid().to_string(),
                logstat: logstat, 
                facility: log_facility,
                logmask: 0xff,
                stream: sock,
                _p: PhantomData,
                _p2: PhantomData,
            }
        );
    }

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

        /// Returns the type of the socket.
    #[inline]
    pub(crate)
    fn get_taptype(&self) -> TapType
    {
        return self.stream.get_type();
    }

    /* 
    /// Returns the maximum msg size in bytes. (Full msg with headers.)
    #[inline]
    pub(crate) 
    fn get_max_msg_size(&self) -> usize
    {
        return self.stream.get_max_msg_size();
    }
    */

    #[inline]
    pub(crate) 
    fn set_logtag<L: AsRef<str>>(&mut self, logtag: L, update_pid: bool)
    {
        self.logtag = 
            truncate_n(logtag.as_ref(), RFC_MAX_APP_NAME).to_string();

        if update_pid == true
        {
            self.logpid = portable::get_pid().to_string();
        }

        return;
    }

    /// Disconnects the unix stream from syslog.
    #[inline]
    pub(crate) async 
    fn disconnectlog(&mut self)  -> SyRes<()>
    {
        return 
            self
                .stream
                .disconnectlog()
                .await
                .map_err(|e| map_error_os!(e, "can not disconnect log properly"));
    }
}

impl<F: SyslogFormatter + Send, D: AsyncSyslogDestination, IO: AsyncSyslogInternalIO> AsyncSyslogApi<F, D>
for AsyncSyslogInternal<F, D, IO>
{
    async 
    fn update_tap_data(&mut self, tap_data: D) -> SyRes<()>
    {
        let is_con = self.stream.is_connected();

        if is_con == true
        {
            self
                .stream
                .disconnectlog()
                .await
                .map_err(|e|
                    map_error_os!(e, "update_tap_data() can not disconnect log properly")
                )?;
        }

        self.stream.update_tap_data(tap_data);

        if is_con == true
        {
            // replace with new instance
            self.stream.connectlog().await?;
        }

        return Ok(());
    }
 
    #[inline]
    fn change_identity(&mut self, ident: &str)
    {
        self.set_logtag(ident, true);
    }

    async 
    fn reconnect(&mut self) -> SyRes<()>
    {
        self.disconnectlog().await?;

        self.connectlog().await?;

        return Ok(());
    }

    async 
    fn closelog(&mut self) -> SyRes<()>
    {
        return self.disconnectlog().await;
    }
    
    fn set_logmask(&mut self, logmask: i32) -> i32
    {
        let oldmask = self.logmask;

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

        return oldmask;
    }


    /// Connects unix stream to the syslog and sets up the properties of
    /// the unix stream.
    #[inline]
    async 
    fn connectlog(&mut self) -> SyRes<()>
    {
        return self.stream.connectlog().await;
    }
    
    /// An internal function which is called by the syslog or vsyslog.
    async 
    fn vsyslog1(&mut self, pri: Priority, fmt: F)
    {
        /*
        // check for invalid bits
        if let Err(e) = pri.check_invalid_bits()
        {
            IO::send_to_stderr(self.logstat, &e.to_string()).await;
        }
        */

        /*match check_invalid_bits(&mut pri)
        {
            Ok(_) => {},
            Err(_e) => self.vsyslog1(get_internal_log(), fmt).await
        }*/

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

        // set default facility if not specified in pri
        let pri_fac = SyslogMsgPriFac::set_facility(pri, self.facility);

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

        let mut msg_formatted = 
            fmt.vsyslog1_format(D::SocketTap::get_max_msg_size(), pri_fac, &self.logtag, msg_pid);

        // output to stderr if required
        IO::send_to_stderr(self.logstat, msg_formatted.get_stderr_output()).await;

        if self.stream.is_connected() == false
        {
            // open connection
            match self.connectlog().await
            {
                Ok(_) => {},
                Err(e) =>
                {
                    IO::send_to_stderr(self.logstat, &e.into_inner() ).await;
                    return;
                }
            }
        }
        
        let fullmsg = msg_formatted.get_full_msg();


        // 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

        loop
        {
            match self.stream.send(fullmsg.as_bytes()).await
            {
                Ok(_) => 
                    return,
                Err(err) =>
                {   
                    if self.get_taptype().is_network() == false
                    {
                        #[cfg(target_family = "unix")]
                        {
                            if let Some(nix::libc::ENOBUFS) = err.raw_os_error()
                            {
                                // scenario 2
                                if self.get_taptype().is_priv() == true
                                {
                                    break;
                                }

                                IO::sleep_micro(1).await;
                                //sleep(Duration::from_micros(1)).await;
                            }
                            else
                            {
                                // scenario 1
                                let _ = self.disconnectlog().await;
                                match self.connectlog().await
                                {
                                    Ok(_) => {},
                                    Err(_e) => break,
                                }

                               
                            } 
                        }

                         #[cfg(target_family = "windows")]
                        {
                            let Ok(werr) = err.downcast::<windows::core::Error>()
                                else
                                {
                                    IO::send_to_stderr(self.logstat, "error downcast failed").await;
                                    break;
                                };

                            IO::send_to_stderr(self.logstat, &werr.message()).await;   
              
                            break;
                        }

                        // if resend will fail then probably the scn 2 will take place
                    }
                    else
                    {
                        let _ = self.disconnectlog().await;
                        match self.connectlog().await
                        {
                            Ok(_) => {},
                            Err(_e) => break,
                        }
                    }  
                }
            }
        } // loop


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

        IO::send_to_syscons(self.logstat, msg_formatted.get_stderr_output()).await;
    }
}