syslog_rs/sync/

syslog_sync_queue.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::borrow::Cow;
use std::marker::PhantomData;
use std::{fmt, thread};
use std::sync::{Arc, Mutex, Weak};
use std::sync::atomic::{AtomicBool, Ordering};

use crate::formatters::{DefaultSyslogFormatter, SyslogFormatter};
use crate::{map_error, SyslogDestination, SyslogLocal};
use crate::common::*;
use crate::error::SyRes;

use super::{syslog_trait::SyslogApi, StreamableSyslogApi, syslog_sync_internal::SyncSyslogInternal, SyslogStream, StreamableSyslog};


/// A wrapper for the data commands in the queue
pub enum SyCmd<F: SyslogFormatter, D: SyslogDestination, S: SyslogQueueChannel<F, D>>
{
    /// A message to syslog server
    Syslog
    {
        pri: Priority,
        msg: F
    },

    /// A reuest to change logmask
    Logmask
    {
        logmask: i32, 
        loopback: S::OneShotChannelSnd<i32>,
    },

    /// A request to change identity
    ChangeIdentity
    {
        identity: String,
    },

    /// Updates the tap settings
    UpdateTap
    {
        tap_type: D,//TapTypeData,
        loopback: S::OneShotChannelSnd<SyRes<()>>,
    },

    ConnectLog
    {
        loopback: S::OneShotChannelSnd<SyRes<()>>
    },

    DisconnectLog
    {
        loopback: S::OneShotChannelSnd<SyRes<()>>
    },

    /// A request to rotate file or reconnect.
    Reconnect,

    /// A request to stop processing and quit
    #[allow(unused)]
    Stop,
}


impl<F: SyslogFormatter, D: SyslogDestination, S: SyslogQueueChannel<F, D>> SyCmd<F, D, S>
{
    /// Construct a message with data to send.
    pub(crate) 
    fn form_syslog(pri: Priority, msg: F) -> Self
    {
        return 
            Self::Syslog
            {
                pri, msg
            };
    }

    pub(crate) 
    fn form_connectlog() -> (Self, S::OneShotChannelRcv<SyRes<()>>)
    {
        let (tx, rx) = S::create_oneshot_channel::<SyRes<()>>();

        return 
            (Self::ConnectLog{ loopback: tx }, rx);
    }

    pub(crate) 
    fn form_disconnectlog() -> (Self, S::OneShotChannelRcv<SyRes<()>>)
    {
        let (tx, rx) = S::create_oneshot_channel::<SyRes<()>>();

        return 
            (Self::DisconnectLog{ loopback: tx }, rx);
    }

    /// Constructs a message to make logmask with or without previous PRI.
    /// 
    /// # Arguments
    /// 
    /// * `logmask` - a new logmask
    pub(crate) 
    fn form_logmask(logmask: i32) -> (Self, S::OneShotChannelRcv<i32>)
    {
        let (tx, rx) = S::create_oneshot_channel::<i32>();

        return 
            (Self::Logmask{ logmask, loopback: tx }, rx);
    }

    /// Constructs a message which should change the identity (appname) of the
    /// instance.
    pub(crate) 
    fn form_change_ident(identity: String) -> Self
    {
        return 
            Self::ChangeIdentity
            {
                identity: identity
            };
    }

    /// Constructs a message which changes the destination of the log messages i.e
    /// changing path of the dst file or address. The `new_tap_type` 
    /// should be the same variant [TapTypeData] as previous.
    pub(crate)
    fn form_update_tap(new_tap_type: D/*TapTypeData*/) -> (Self, S::OneShotChannelRcv<SyRes<()>>)
    {
        let (tx, rx) = S::create_oneshot_channel::<SyRes<()>>();

        return (
            Self::UpdateTap  
            { 
                tap_type: new_tap_type,
                loopback: tx
            },
            rx
        );
    }

    /// Constructs a message to handle SIGHUP. This is usefull only when the instance
    /// is writing directly into the file. Or just reconnect.
    pub(crate) 
    fn form_reconnect() -> Self
    {
        return Self::Reconnect;
    }

    /// Constructs a message to stop thread gracefully. After receiving this
    /// message a thread will quit and all messages that would be sent after
    /// this message will be cleared from queue and a new messages will not be
    /// received.
    #[allow(unused)]
    pub(crate) 
    fn form_stop() -> Self
    {
        return Self::Stop;
    }
}

/// Internal struct of the syslog client thread.
struct SyslogInternal<F: SyslogFormatter, D: SyslogDestination, S: SyslogQueueChannel<F, D>>
{
    /// A explicit stop flag
    run_flag: Arc<AtomicBool>,

    /// commands channel
    tasks: S::ChannelRcv,

    /// An syslog assets
    inner: SyncSyslogInternal<F, D>,
}



impl<F, D, S> SyslogInternal<F, D, S>
where F: SyslogFormatter, D: SyslogDestination, S: SyslogQueueChannel<F, D>
{
    fn new(logstat: LogStat, ssi: SyncSyslogInternal<F, D>) -> SyRes<(Self, S::ChannelSnd, Weak<AtomicBool>)>
    {
        // control flag
        let run_flag: Arc<AtomicBool> = Arc::new(AtomicBool::new(true));
        let run_control = Arc::downgrade(&run_flag);

        // creating queue for messages
        let (sender, receiver) = S::create_channel();

        // creating internal syslog struct
        let mut inst = 
            SyslogInternal
            {
                run_flag: run_flag,
                tasks: receiver,
                inner: ssi,
            };

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

        return Ok((inst, sender, run_control));
    }

    fn thread_worker(mut self)
    {
        loop
        {
            // self will be dropped as soon as thread will be stopped
            if self.run_flag.load(Ordering::Relaxed) == false
            {
                // force leave
                break;
            }	

            match self.tasks.q_recv_blocking()
			{
				Some(task) =>
				{
                    match task
                    {
                        SyCmd::Syslog{ pri, msg } =>
                        {
                            self.inner.vsyslog1(pri, &msg);
                        },
                        SyCmd::Logmask{ logmask, loopback } =>
                        {
                            let pri = self.inner.set_logmask(logmask);

                            let _ = loopback.send_once_blocking(pri);
                        },
                        SyCmd::ChangeIdentity{ identity } =>
                        {
                            self.inner.set_logtag(identity, true);
                        },

                        SyCmd::UpdateTap{ tap_type, loopback } =>
                        {
                            if self.inner.get_taptype().is_file() == true
                            {
                                let res = self.inner.update_tap_data(tap_type);
                                
                                if let Err(Err(e)) = loopback.send_once_blocking(res)
                                {
                                    self.inner.send_to_stderr(&[Cow::Owned(e.to_string())]);
                                }
                            }
                        },

                        SyCmd::ConnectLog{ loopback} => 
                        {
                            if let Err(Err(e)) = loopback.send_once_blocking(self.inner.connectlog())
                            {
                                self.inner.send_to_stderr(&[Cow::Owned(e.to_string())]);
                            }
                        },

                        SyCmd::DisconnectLog{ loopback} => 
                        {
                            if let Err(Err(e)) = loopback.send_once_blocking(self.inner.disconnectlog())
                            {
                                self.inner.send_to_stderr(&[Cow::Owned(e.to_string())]);
                            }
                        },

                        SyCmd::Reconnect =>
                        {
                            if let Err(e) = self.inner.disconnectlog()
                            {
                                self.inner.send_to_stderr(&[Cow::Owned(e.to_string())]);
                            }
                                
                            if let Err(e) = self.inner.connectlog()
                            {
                                self.inner.send_to_stderr(&[Cow::Owned(e.to_string())]);
                            }
                        },
                        SyCmd::Stop =>
                        {
                            // ignore the rest
                            break;
                        }
                    }
                },
                None =>
                {
                    break;
                }
            } // match

        } // loop

        return;
    }
}

/// A trait which should be implemented by the channel provider which forms a command queue.
/// This trait provides a blocking receive interface on the receiver side.
pub trait SyslogQueueChanRcv<F: SyslogFormatter, D: SyslogDestination, S: SyslogQueueChannel<F, D>>: fmt::Debug + Send
{
    /// Receive from channel in blocking mode.
    fn q_recv_blocking(&mut self) -> Option<SyCmd<F, D, S>>;
}

/// A trait which should be implemented by the channel provider which forms a command queue.
/// This trait provides both or either the blocking send interface.
#[allow(async_fn_in_trait)]
pub trait SyslogQueueChanSnd<F: SyslogFormatter, D: SyslogDestination, S: SyslogQueueChannel<F, D>>: fmt::Debug + Clone
{
    /// Sends the message over the channel in blocking mode.
    fn q_send_blocking(&self, msg: SyCmd<F, D, S>) -> SyRes<()>;

    /// Sends the message over the channel in async mode. By default, it returns error.
    async fn q_send(&self, _msg: SyCmd<F, D, S>) -> SyRes<()>
    {
        crate::throw_error!("async is not availabe here"); 
    }
}

/// A trait which should be implemented by the channel provider which forms a command queue.
/// This trait provides a Oneshot channel a receive interface for both sync and async modes.
#[allow(async_fn_in_trait)]
pub trait SyslogQueueOneChanRcv<C>
{
    /// Receives once from the channel in sync mode.
    fn recv_once_blocking(self) -> SyRes<C>;

    /// Receives once from the channel in async mode. By default, it returns error if not 
    /// implemented.
    async fn recv_once(self) -> SyRes<C> where Self: Sized
    {
        crate::throw_error!("async is not availabe here"); 
    }
}

/// A trait which should be implemented by the channel provider which forms a command queue.
/// This trait provides a Oneshot channel a send interface for blocking mode.
pub trait SyslogQueueOneChanSnd<C: Send>: fmt::Debug + Send
{
    /// Send to channel in blocing mode.
    fn send_once_blocking(self, data: C) -> Result<(), C>;
}

/// A trait which should be implemented by the channel provider which forms a command queue.
/// This trait provides a common interface which includes everything i.e channel, oneshot 
/// channel and manipulations.
pub trait SyslogQueueChannel<F: SyslogFormatter, D: SyslogDestination>: fmt::Debug + Send + Clone + 'static
{
    /// A send side of the channel type.
    type ChannelSnd: SyslogQueueChanSnd<F, D, Self>;
    
    /// A receive side of the channel type.
    type ChannelRcv: SyslogQueueChanRcv<F, D, Self>;

    /// A oneshot send side of the channel type.
    type OneShotChannelSnd<C: Send + fmt::Debug>: SyslogQueueOneChanSnd<C>;

    /// A oneshor receive side of the channel type.
    type OneShotChannelRcv<C>: SyslogQueueOneChanRcv<C>;

    /// Creates unbounded channel.
    fn create_channel() -> (Self::ChannelSnd, Self::ChannelRcv);   

    /// Creates oneshot channel.
    fn create_oneshot_channel<C: Send + fmt::Debug>() -> (Self::OneShotChannelSnd<C>, Self::OneShotChannelRcv<C>);
}



/// An instance which is shared between all threads, but a MPSC channel is
/// used to manage the simultanious action. A `thread worker` is receiving 
/// messages, formats and forwards to syslog serer (destination).
/// 
/// For this isntance a [SyslogApi] and [StreamableSyslogApi] are implemented.
/// 
/// Also if `async` is enabled, a [AsyncSyslogQueueApi] is implemented.
/// 
/// # Generics
/// 
/// * `S` - a [SyslogQueueChannel] a MPSC provider.
/// 
/// * `F` - a [SyslogFormatter] which sets the instance which would 
///     format the message.
/// 
/// * `D` - a [SyslogDestination] instance which is either:
///     [SyslogLocal], [SyslogFile], [SyslogNet], [SyslogTls]. By
///     default a `SyslogLocal` is selected.
#[derive(Debug, Clone)]
pub struct SyslogQueue<S, F = DefaultSyslogFormatter, D = SyslogLocal>
where F: SyslogFormatter, D: SyslogDestination, S: SyslogQueueChannel<F, D>
{   
    /// Control flag
    run_control: Weak<AtomicBool>,

    /// commands channel
    pub(crate) tasks: S::ChannelSnd,//q_adapter::SendChannelSyCmd<D>,

    /// process thread
    thread: Arc<Mutex<Option<thread::JoinHandle<()>>>>,

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

    /// phantom for [SyslogDestination]
    _p2: PhantomData<D>,
}

unsafe impl<F, D, S> Send for SyslogQueue<S, F, D> 
where F: SyslogFormatter, D: SyslogDestination, S: SyslogQueueChannel<F, D>
{}

unsafe impl<F, D, S>  Sync for SyslogQueue<S, F, D> 
where F: SyslogFormatter, D: SyslogDestination, S: SyslogQueueChannel<F, D>
{}

impl<F, D, S> Drop for SyslogQueue<S, F, D>
where F: SyslogFormatter, D: SyslogDestination, S: SyslogQueueChannel<F, D>
{
    fn drop(&mut self) 
    {
        if let Some(ctrl) = self.run_control.upgrade()
        {   
            ctrl.store(false, Ordering::SeqCst);

            if let Err(_e) = self.tasks.q_send_blocking(SyCmd::form_stop())
            {

            }

            let join_handle = self.thread.lock().unwrap().take().unwrap();

            let _ = join_handle.join();
        }
    }
}

impl<S: SyslogQueueChannel<DefaultSyslogFormatter, SyslogLocal>> SyslogQueue<S>
{

    pub 
    fn openlog(ident: Option<&str>, logstat: LogStat, facility: LogFacility, net_tap: SyslogLocal) -> SyRes<Self>
    {
        // creating internal syslog struct

        let (inst, sender, run_ctrl) = 
            SyslogInternal
                ::<DefaultSyslogFormatter, SyslogLocal, S>
                ::new(
                    logstat,
                    SyncSyslogInternal::<DefaultSyslogFormatter, _>::new(ident, logstat, facility, net_tap)?
                )?;
        
        
        let thr_name: String = "syslog_queue/0".into();

        // initiate a thread
        let thread_hnd = 
            thread::Builder::new()
                .name(thr_name.clone())
                .spawn(move || SyslogInternal::<DefaultSyslogFormatter, SyslogLocal, S>::thread_worker(inst))
                .map_err(|e| 
                    map_error!("{} thread spawn failed. {}", thr_name, e)
                )?;

        // creating a syslog public struct instance
        let ret = 
            Self
            {
                run_control: run_ctrl,
                tasks: sender,
                thread: Arc::new(Mutex::new(Some(thread_hnd))),
                _p: PhantomData::<DefaultSyslogFormatter>,
                _p2: PhantomData::<SyslogLocal>,
            };

        return Ok(ret);
    }
}

impl<F, D, S> SyslogQueue<S, F, D>
where F: SyslogFormatter, D: SyslogDestination, S: SyslogQueueChannel<F, D>
{
    pub
    fn openlog_with(ident: Option<&str>, logstat: LogStat, facility: LogFacility, net_tap: D) -> SyRes<SyslogQueue<S, F, D>>
    {
        // creating internal syslog struct

        let (inst, sender, run_ctrl) = 
            SyslogInternal
                ::<F, D, S>
                ::new(
                    logstat,
                    SyncSyslogInternal::<F, D>::new(ident, logstat, facility, net_tap)?
                )?;
        
        
        let thr_name: String = "syslog_queue/0".into();

        // initiate a thread
        let thread_hnd = 
            thread::Builder::new()
                .name(thr_name.clone())
                .spawn(move || SyslogInternal::<F, D, S>::thread_worker(inst))
                .map_err(|e| 
                    map_error!("{} thread spawn failed. {}", thr_name, e)
                )?;

        // creating a syslog public struct instance
        let ret = 
            Self
            {
                run_control: run_ctrl,
                tasks: sender,
                thread: Arc::new(Mutex::new(Some(thread_hnd))),
                _p: PhantomData::<F>,
                _p2: PhantomData::<D>,
            };

        return Ok(ret);
    }
}

impl<F: SyslogFormatter, D: SyslogDestination, S: SyslogQueueChannel<F, D>> StreamableSyslogApi<D, F, SyslogQueue<S, F, D>> 
for SyslogQueue<S, F, D>
{
    fn make_stream(&self, pri: Priority) -> Box<dyn SyslogStream> 
    {
        let inst = 
            StreamableSyslog
            {
                inner: self.clone(),
                pri: pri,
                _p: PhantomData::<F>,
                _p1: PhantomData::<D>,

            };

        return Box::new(inst);
    }
}

impl<F: SyslogFormatter, D: SyslogDestination, S: SyslogQueueChannel<F, D>> SyslogApi<F, D> 
for SyslogQueue<S, F, D>
{
    fn connectlog(&self) -> SyRes<()>
    {
        let (sy_cmd, loopback) = 
            SyCmd::form_connectlog();

        self.tasks.q_send_blocking(sy_cmd)?;

        return 
            loopback
                .recv_once_blocking()?;
    }

    fn setlogmask(&self, logmask: i32) -> SyRes<i32> 
    {
        let (sy_cmd, loopback) = 
            SyCmd::form_logmask(logmask);

        self.tasks.q_send_blocking(sy_cmd)?;

        return 
            loopback
                .recv_once_blocking();
    }

    fn closelog(&self) -> SyRes<()> 
    {
        let (sy_cmd, loopback) = 
            SyCmd::form_disconnectlog();

        // send stop
        self.tasks.q_send_blocking(sy_cmd)?;

        return 
            loopback
                .recv_once_blocking()?;
    }

    fn syslog(&self, pri: Priority, fmt: F) 
    {
        // even if the thread is in a process of termination, there is
        // no need to sync access to the run_control field as even if
        // syslog thread will terminate before someone push something on the
        // queue, it will be left in the queue until the end of program's time.
        
        let sy_cmd = SyCmd::form_syslog(pri, fmt);

        let _ = self.tasks.q_send_blocking(sy_cmd);
        //tasks.send(sy_cmd);

        return;
    }

    fn change_identity(&self, ident: &str) -> SyRes<()>
    {
        let sy_cmd = 
            SyCmd::form_change_ident(ident.to_string());

        return 
            self.tasks.q_send_blocking(sy_cmd);
    }

    fn reconnect(&self) -> SyRes<()>
    {
        return 
            self.tasks.q_send_blocking(SyCmd::form_reconnect());
    }

    fn update_tap_data(&self, tap_data: D) -> SyRes<()>
    {
        let (tap_data_cmd, loopback) = SyCmd::form_update_tap(tap_data);

        self.tasks.q_send_blocking(tap_data_cmd)?;

        return 
            loopback
                .recv_once_blocking()?;
    }
}

/// A queued implementation for ASYNC.
#[cfg(all(feature = "build_with_queue", feature = "async_enabled"))]
pub mod syslog_async_queue
{
    use crate::error::SyRes;
    use crate::sy_sync_queue::{SyCmd, SyslogQueueChanSnd, SyslogQueueChannel, SyslogQueueOneChanRcv};
    use crate::Priority;
    use crate::{formatters::SyslogFormatter, SyslogDestination, SyslogQueue};
    use crate::a_sync::syslog_trait::AsyncSyslogQueueApi;

    impl<F: SyslogFormatter, D: SyslogDestination, S: SyslogQueueChannel<F, D>> AsyncSyslogQueueApi<F, D> 
    for SyslogQueue<S, F, D>
    {
        async 
        fn a_connectlog(&mut self) -> SyRes<()> 
        {
            let (sy_cmd, loopback) = 
                SyCmd::form_connectlog();

            self.tasks.q_send(sy_cmd).await?;

            return 
                loopback
                    .recv_once()
                    .await?;
        }

        /// Sets the logmask to filter out the syslog calls. This function behaves 
        /// differently as it behaves in syslog_sync.rs or syslog_async.rs.
        /// It may return an error if: syslog thread had exit and some thread calls
        /// this function. Or something happened with channel. 
        /// This function blocks until the previous mask is received.
        /// 
        /// See macroses [LOG_MASK] and [LOG_UPTO] to generate mask
        ///
        /// # Example
        ///
        /// LOG_MASK!(Priority::LOG_EMERG) | LOG_MASK!(Priority::LOG_ERROR)
        ///
        /// or
        ///
        /// ~(LOG_MASK!(Priority::LOG_INFO))
        /// LOG_UPTO!(Priority::LOG_ERROR) 
        async 
        fn a_setlogmask(&self, logmask: i32) -> SyRes<i32>
        {
            let (sy_cmd, loopback) = 
                SyCmd::form_logmask(logmask);

            self.tasks.q_send(sy_cmd).await?;

            return 
                loopback
                    .recv_once()
                    .await;
        }

        /// Closes connection to the syslog server
        async 
        fn a_closelog(&self) -> SyRes<()>
        {
            let (sy_cmd, loopback) = 
                SyCmd::form_disconnectlog();

            // send stop
            self.tasks.q_send(sy_cmd).await?;

            return 
                loopback
                    .recv_once()
                    .await?;
        }

        /// Similar to libc, syslog() sends data to syslog server, but asynchroniously.
        /// 
        /// # Arguments
        ///
        /// * `pri` - a priority [Priority]
        ///
        /// * `fmt` - a program's message to be sent as payload. The message is encoded with the
        ///     [SyslogFormatter] and may be different for different formatters.
        #[inline]
        async 
        fn a_syslog(&self, pri: Priority, fmt: F)
        {
            let sy_cmd = SyCmd::form_syslog(pri, fmt);

            let _ = self.tasks.q_send(sy_cmd).await;

            return;
        }


        /// Performs the reconnection to the syslog server or file re-open.
        /// 
        /// # Returns
        /// 
        /// A [Result] is retured as [SyRes].
        /// 
        /// * [Result::Ok] - with empty inner type.
        /// 
        /// * [Result::Err] - an error code and description
        async 
        fn a_reconnect(&self) -> SyRes<()>
        {
            return 
                self.tasks.q_send(SyCmd::form_reconnect()).await;
        }

        async 
        fn a_change_identity(&self, ident: &str) -> SyRes<()> 
        {
            let sy_cmd = 
                SyCmd::form_change_ident(ident.to_string());

            return 
                self.tasks.q_send(sy_cmd).await;
        }

        /// Updates the inner instance destionation i.e path to file
        /// or server address. The type of destination can not be changed.
        /// 
        /// This function disconnects from syslog server if previously was 
        /// connected (and reconnects if was connected previously).
        /// 
        /// # Arguments 
        /// 
        /// * `new_tap` - a consumed instance of type `D` [SyslogDestination]
        /// 
        /// # Returns 
        /// 
        /// A [SyRes] is returned. An error may be returned if:
        /// 
        /// * connection to server was failed
        /// 
        /// * incorrect type
        /// 
        /// * disconnect frm server failed
        async 
        fn a_update_tap_data(&self, new_tap: D) -> SyRes<()>
        {
            let (tap_data_cmd, loopback) = SyCmd::form_update_tap(new_tap);

            self.tasks.q_send(tap_data_cmd).await?;

            return 
                loopback
                    .recv_once()
                    .await?;
        }
    }
}