syslog_rs/sync/

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

use tokio::sync::mpsc::{unbounded_channel, UnboundedReceiver, UnboundedSender};
use tokio::sync::oneshot;


use crate::formatters::SyslogFormatter;
use crate::socket::TapTypeData;
use crate::{map_error, map_error_code, throw_error};
use crate::common::*;
use crate::error::SyRes;


use super::syslog_sync_internal::SyncSyslogInternal;
use super::syslog_trait::SyslogApi;

#[derive(Debug)]
pub struct SyslogQueueAdapter
{
    /// A control status check.
    run_control: Weak<AtomicBool>,

    /// An incoming messages.
    tx: UnboundedSender<SyCmd>
}

impl SyslogQueueAdapter
{
    fn new(tx: UnboundedSender<SyCmd>, run_control:  Weak<AtomicBool>) -> Self
    {
        return Self{ run_control, tx };
    }

    pub(crate) 
    fn consume(self) -> SyRes<UnboundedSender<SyCmd>>
    {
        if let None = self.run_control.upgrade()
        {
            throw_error!("Can not attach queue adapter to async, a queue thread is not running!");
        }

        return Ok(self.tx);
    }
}

/// A wrappes for the data in the queue
pub(crate) enum SyCmd
{
    /// A message to syslog server
    Syslog
    {
        pri: Priority,
        msg: String
    },

    /// A reuest to change logmask
    Logmask
    {
        logmask: i32, 
        loopback: oneshot::Sender<i32>,
    },

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

    /// Updates the tap settings
    UpdateTap
    {
        tap_type: TapTypeData,
        loopback: oneshot::Sender<SyRes<()>>,
    },

    ConnectLog
    {
        loopback: oneshot::Sender<SyRes<()>>
    },

    DisconnectLog
    {
        loopback: oneshot::Sender<SyRes<()>>
    },

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

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

impl SyCmd
{
    /// Construct a message with data to send.
    pub(crate) 
    fn form_syslog(pri: Priority, msg: String) -> Self
    {
        return 
            Self::Syslog
            {
                pri, msg
            };
    }

    pub(crate) 
    fn form_connectlog() -> (Self, oneshot::Receiver<SyRes<()>>)
    {
        let (tx, rx) = oneshot::channel::<SyRes<()>>();

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

    pub(crate) 
    fn form_disconnectlog() -> (Self, oneshot::Receiver<SyRes<()>>)
    {
        let (tx, rx) = 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, oneshot::Receiver<i32>)
    {
        let (tx, rx) = 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: TapTypeData) -> (Self, oneshot::Receiver<SyRes<()>>)
    {
        let (tx, rx) = 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;
    }
}

struct SyslogInternal<F: SyslogFormatter>
{
    /// A explicit stop flag
    run_flag: Arc<AtomicBool>,

    /// commands channel
    tasks: UnboundedReceiver<SyCmd>,

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



impl<F: SyslogFormatter> SyslogInternal<F>
{
    fn new(logstat: LogStat, ssi: SyncSyslogInternal<F>) -> SyRes<(Self, UnboundedSender<SyCmd>, 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) = unbounded_channel::<SyCmd>();

        // 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.blocking_recv()
			{
				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(pri);
                        },
                        SyCmd::ChangeIdentity{ identity } =>
                        {
                            self.inner.set_logtag(identity, true);
                        },

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

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

                        SyCmd::DisconnectLog{ loopback} => 
                        {
                            if let Err(Err(e)) = loopback.send(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 common instance which describes the syslog state
#[derive(Debug, Clone)]
pub struct SyslogQueue<F: SyslogFormatter>
{   
    /// Control flag
    run_control: Weak<AtomicBool>,

    /// commands channel
    tasks: UnboundedSender<SyCmd>,

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

    _p: PhantomData<F>,
}

unsafe impl<F: SyslogFormatter> Send for SyslogQueue<F> {}
unsafe impl<F: SyslogFormatter> Sync for SyslogQueue<F> {}

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

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

            }

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

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

impl<F: SyslogFormatter> SyslogQueue<F>
{
    pub 
    fn openlog(ident: Option<&str>, logstat: LogStat, facility: LogFacility, net_tap: TapTypeData) -> SyRes<SyslogQueue<F>>
    {
        // creating internal syslog struct

        let (inst, sender, run_ctrl) = 
            SyslogInternal
                ::<F>
                ::new(
                    logstat,
                    SyncSyslogInternal::<F>::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>::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>,
            };

        return Ok(ret);
    }
}

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

        self
            .tasks
            .send(sy_cmd)
            .map_err(|e| map_error_code!(SendError, "connectlog() error: '{}'", e))?;

        return 
            loopback
                .blocking_recv()
                .map_err(|e| 
                    map_error_code!(UnboundedChannelError, "channel error: '{}'", e)
                )?;
    }

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

        self
            .tasks
            .send(sy_cmd)
            .map_err(|e| 
                map_error_code!(SendError, "closelog() error: '{}'", e)
            )?;

        return 
            loopback
                .blocking_recv()
                .map_err(|e| 
                    map_error_code!(UnboundedChannelError, "channel error: '{}'", e)
                );
    }

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

        // send stop
        self
            .tasks
            .send(sy_cmd)
            .map_err(|e| 
                map_error_code!(SendError, "closelog() error: '{}'", e)
            )?;

        return 
            loopback
                .blocking_recv()
                .map_err(|e| 
                    map_error_code!(UnboundedChannelError, "channel error: '{}'", e)
                )?;
    }

    fn syslog(&self, pri: Priority, fmt: String) 
    {
        // 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.send(sy_cmd);

        return;
    }

    fn vsyslog<S: AsRef<str>>(&self, pri: Priority, fmt: S) 
    {
        let sy_cmd = 
            SyCmd::form_syslog(pri, fmt.as_ref().to_string());

        let _ = self.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
                .send(sy_cmd)
                .map_err(|e| 
                    map_error_code!(SendError, "change_identity() error: '{}'", e)
                );
    }

    fn reconnect(&self) -> SyRes<()>
    {
        return 
            self
                .tasks
                .send(SyCmd::form_reconnect())
                .map_err(|e| 
                    map_error_code!(SendError, "signal_rotate_log() error: '{}'", e)
                );
    }

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

        self
            .tasks
            .send(tap_data_cmd)
            .map_err(|e| 
                map_error_code!(SendError, "signal_rotate_log() error: '{}'", e)
            )?;

        return 
            loopback
                .blocking_recv()
                .map_err(|e| 
                    map_error_code!(UnboundedChannelError, "channel error: '{}'", e)
                )?;
    }
}

impl<F: SyslogFormatter> SyslogQueue<F>
{

    pub
    fn make_adapter(&self) -> SyslogQueueAdapter
    {
        return SyslogQueueAdapter::new(self.clone_task_channel(), self.clone_run_ctrl());
    }

    fn clone_run_ctrl(&self) -> Weak<AtomicBool>
    {
        return self.run_control.clone();
    } 

    fn clone_task_channel(&self) -> UnboundedSender<SyCmd>
    {
        return self.tasks.clone();
    }
}