syslog_rs/a_sync/

syslog_async_internal.rs

/*-
 * syslog-rs - a syslog client translated from libc to rust
 * 
 * Copyright 2025 Aleksandr Morozov
 * 
 * Licensed under the EUPL, Version 1.2 or - as soon they will be approved by
 * the European Commission - subsequent versions of the EUPL (the "Licence").
 * 
 * You may not use this work except in compliance with the Licence.
 * 
 * You may obtain a copy of the Licence at:
 * 
 *    https://joinup.ec.europa.eu/software/page/eupl
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the Licence is distributed on an "AS IS" basis, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * Licence for the specific language governing permissions and limitations
 * under the Licence.
 */


use std::borrow::Cow;
use std::marker::PhantomData;

use tokio::time::{sleep, Duration};

use nix::libc;

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::socket::TapTypeData;

use super::async_socket::*;

/// Internal structure with syslog setup
#[derive(Debug)]
pub struct AsyncSyslogInternal<F: SyslogFormatter + Send>
{
    /// 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: Box<dyn AsyncSyslogTap + Send + Sync>,

    _p: PhantomData<F>,
}

//unsafe impl Sync for SyslogInternal{}
//unsafe impl Send for SyslogInternal{}

// Drop is called when no more references are left.
/*
impl Drop for AsyncSyslogInternal 
{
    fn drop(&mut self) 
    {
        // requires await, so skip.
        //self.disconnectlog();
    }
}
    */

impl<F: SyslogFormatter + Send> AsyncSyslogInternal<F>
{
    /// 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 [TapTypeData].
    /// 
    /// # 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: TapTypeData
    ) -> 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()
            };

        return Ok(
            Self
            {
                logtag: logtag,
                logpid: portable::get_pid().to_string(),
                logstat: logstat, 
                facility: log_facility,
                logmask: 0xff,
                stream: AsyncTap::<()>::new(req_tap)?,
                _p: PhantomData
            }
        );
    }

    pub(crate) async 
    fn update_tap_data(&mut self, tap_data: TapTypeData) -> 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(());
    }

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

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

        return oldmask;
    }
 
    #[inline]
    async 
    fn send_to_stderr(&self, msg: &[Cow<'_, str>])
    {
        if self.logstat.intersects(LogStat::LOG_PERROR) == true
        {
            let stderr_lock = tokio::io::stderr();

            let newline = "\n";
            let _ = async_send_to_fd(stderr_lock, msg, newline).await;
        }
    }

    #[inline]
    async 
    fn send_to_syscons(&self, msg_payload: &[Cow<'_, str>])
    {
        use tokio::fs::File;

        if self.logstat.intersects(LogStat::LOG_CONS)
        {
            let syscons = 
                File
                    ::options()
                        .create(false)
                        .read(false)
                        .write(true)
                        .custom_flags(libc::O_NONBLOCK | libc::O_CLOEXEC)
                        .open(*PATH_CONSOLE)
                        .await;

            if let Ok(file) = syscons
            {
                let newline = "\n";
                let _ = async_send_to_fd(file, msg_payload, newline);
            }
        }
    }

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

        return false;
    }

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

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

    /// Connects unix stream to the syslog and sets up the properties of
    /// the unix stream.
    #[inline]
    pub(crate) async 
    fn connectlog(&mut self) -> SyRes<()>
    {
        return self.stream.connectlog().await;
    }
    
    /// An internal function which is called by the syslog or vsyslog.
    pub(crate) async 
    fn vsyslog1(&mut self, mut pri: Priority, fmt: &str)
    {
        // check for invalid bits
        if let Err(e) = pri.check_invalid_bits()
        {
            self.send_to_stderr(&[Cow::Owned(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.bits()) == true
        {
            return;
        }

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

        let progname = self.logtag.clone();
        let pid = self.logpid.clone();

        let msg_formatted = F::vsyslog1_format(self.get_tap_type(), pri, &progname, &pid, fmt);

        // output to stderr if required
        self.send_to_stderr(msg_formatted.get_stderr_output()).await;

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


        // 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_tap_type().is_network() == false
                    {
                        if let Some(libc::ENOBUFS) = err.raw_os_error()
                        {
                            // scenario 2
                            if self.get_tap_type().is_priv() == true
                            {
                                break;
                            }

                            sleep(Duration::from_micros(1)).await;
                        }
                        else
                        {
                            // scenario 1
                            let _ = self.disconnectlog().await;
                            match self.connectlog().await
                            {
                                Ok(_) => {},
                                Err(_e) => 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

        self.send_to_syscons(msg_formatted.get_stderr_output()).await;
    }
}