syslog_rs/sync/

syslog_sync_shared.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::marker::PhantomData;
use std::sync::Arc;
use std::{str};

use instance_copy_on_write::ICoW;

use crate::formatters::DefaultSyslogFormatter;
use crate::sync::syslog_stream::SyStreamApi;
use crate::sync::syslog_sync_internal::{SyslogSocket};
use crate::sync::{LogItems, SyStream};
use crate::{formatters::SyslogFormatter, map_error_code};
use crate::{common::*, SyslogDestination, SyslogLocal};
use crate::error::SyRes;

use super::syslog_trait::SyslogApi;


/// A sync, shared instance of the syslog client which is shared between
/// threads. Previously a mutex was used, but since the v5.0.0 a CoW experimental
/// approach is used.
/// 
/// For this isntance a [SyslogApi] and [SyStreamApi] are implemented.
/// 
/// ```ignore
/// let log = 
///     SyncSyslog::openlog(
///         Some("test1"), 
///         LogStat::LOG_CONS | LogStat::LOG_NDELAY | LogStat::LOG_PID, 
///         LogFacility::LOG_DAEMON,
///         SyslogLocal::new()
///     );
/// ```
/// 
/// ```ignore
/// let log = 
///     SyncSyslog
///         ::<DefaultSyslogFormatter, SyslogLocal>
///         ::openlog_with(
///             Some("test1"), 
///             LogStat::LOG_CONS | LogStat::LOG_NDELAY | LogStat::LOG_PID, 
///             LogFacility::LOG_DAEMON,
///             SyslogLocal::new()
///         );
/// ```
/// 
/// ```ignore
/// pub static SYSLOG3: LazyLock<SyncSyslog<DefaultSyslogFormatter, SyslogLocal,>> = 
///     LazyLock::new(|| 
///         {
///             SyncSyslog
///                 ::<DefaultSyslogFormatter, SyslogLocal>
///                 ::openlog_with(
///                     Some("test1"), 
///                     LogStat::LOG_CONS | LogStat::LOG_NDELAY | LogStat::LOG_PID, 
///                     LogFacility::LOG_DAEMON,
///                     SyslogLocal::new()
///                 )
///                 .unwrap()
///         }
///     );
/// ```
/// 
/// A stream is availble via [SyStreamApi].
/// 
/// ```ignore
/// let _ = write!(SYSLOG.stream(Priority::LOG_DEBUG), "test {} 123 stream test ", d);
/// ```
/// 
/// # Generics
/// 
/// * `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 SyncSyslog<F = DefaultSyslogFormatter, D = SyslogLocal>
where 
    F: SyslogFormatter, 
    D: SyslogDestination, 
{   
    /// An identification i.e program name, thread name
    log_items: Arc<ICoW<LogItems>>,

    /// A stream (unixdatagram, udp, tcp)
    stream: Arc<SyslogSocket<D>>,

     _p: PhantomData<F>,
}

unsafe impl<F: SyslogFormatter, D: SyslogDestination> Send for SyncSyslog<F, D>
{}

impl SyncSyslog
{
    /// Opens a default connection to the local syslog server with default formatter.
    /// 
    /// In order to access the syslog API, use the [SyslogApi].
    /// 
    /// # Arguments
    /// 
    /// * `ident` - A program name which will appear on the logs. If none, will be determined
    ///     automatically.
    /// 
    /// * `logstat` - [LogStat] an instance config.
    /// 
    /// * `facility` - [LogFacility] a syslog facility.
    /// 
    /// * `net_tap_prov` - a [SyslogLocal] instance with configuration.
    /// 
    /// # Returns
    /// 
    /// A [SyRes] is returned ([Result]) with: 
    /// 
    /// * [Result::Ok] - with instance
    /// 
    /// * [Result::Err] - with error description.
    pub 
    fn openlog(ident: Option<&str>, logstat: LogStat, facility: LogFacility, net_tap_prov: SyslogLocal) -> SyRes<Self> 
    {
        return Ok( 
            Self
            {
                log_items: 
                    Arc::new(
                        ICoW::new(
                            LogItems::new(ident, 0xff, logstat, facility)
                        )
                    ),
                stream: 
                    Arc::new(
                        SyslogSocket::<SyslogLocal>::new(logstat, net_tap_prov)?
                    ),
                _p: 
                    PhantomData,
            }
        );
    }
}


impl<F: SyslogFormatter, D: SyslogDestination> SyncSyslog<F, D>
{
    /// Opens a special connection to the destination syslog server with specific formatter.
    /// 
    /// All struct generic should be specified before calling this function.
    /// 
    /// In order to access the syslog API, use the [SyslogApi].
    /// 
    /// # Arguments
    /// 
    /// * `ident` - A program name which will appear on the logs. If none, will be determined
    ///     automatically.
    /// 
    /// * `logstat` - [LogStat] an instance config.
    /// 
    /// * `facility` - [LogFacility] a syslog facility.
    /// 
    /// * `net_tap_prov` - a destination server. A specific `D` instance which contains infomation 
    ///     about the destination server. See `syslog_provider.rs`.
    /// 
    /// # Returns
    /// 
    /// A [SyRes] is returned ([Result]) with: 
    /// 
    /// * [Result::Ok] - with instance
    /// 
    /// * [Result::Err] - with error description.
    pub 
    fn openlog_with(ident: Option<&str>, logstat: LogStat, facility: LogFacility, net_tap_prov: D) -> SyRes<Self> 
    {
        return Ok( 
            Self
            {
                log_items: 
                    Arc::new(
                        ICoW::new(
                            LogItems::new(ident, 0xff, logstat, facility)
                        )
                    ),
                stream: 
                    Arc::new(
                        SyslogSocket::<D>::new(logstat, net_tap_prov)?
                    ),
                _p: 
                    PhantomData,
            }
        );
    }
}


impl<F: SyslogFormatter, D: SyslogDestination> SyslogApi<F, D> for SyncSyslog<F, D>
{
    fn connectlog(&self) -> SyRes<()>
    {
        return 
            self
                .stream
                .connectlog();
    }

    fn setlogmask(&self, logmask: i32) -> SyRes<i32> 
    {
        let mut transaction = 
            self
                .log_items
                .as_ref()
                .clone();
                /*.map_err(|e|
                    map_error_code!(CoWExclLock, "set logmask failed with: {}", e)
                )?;*/

        let pri = 
            transaction
                .set_logmask(logmask);

        transaction
            .commit()
            .map_err(|e|
                map_error_code!(CoWWriteError, "set logmask failed with: {}", e.0)
            )?;

        return Ok(pri);
    }

    fn closelog(&self) -> SyRes<()> 
    {
        return 
            self
                .stream
                .disconnectlog();
    }

    #[inline]
    fn syslog(&self, pri: Priority, fmt: F) 
    {
        let Some((formatted_msg, logstat)) = 
            self.log_items.read().vsyslog1_msg::<F, D>(pri, &fmt)
            else { return };

        self.stream.vsyslog1(logstat, formatted_msg);

        return;
    }

    /// This function can be used to update the facility name, for example
    /// after fork().
    /// 
    /// # Arguments
    /// 
    /// * `ident` - a new identity (up to 48 UTF8 chars)
    fn change_identity(&self, ident: Option<&str>) -> SyRes<()>
    {
        let mut clonned = 
            self
                .log_items
                .as_ref()
                .clone();
                /*.map_err(|e|
                    map_error_code!(CowTransactionFailed, "change_identity to '{:?}' was not completed due to error: '{}'", ident, e)
                )?;*/

        clonned.set_identity(ident);

        clonned
            .commit()
            .map_err(|e|
                map_error_code!(CoWWriteError, "cannot change_identity due to exclusive lock, err: '{}'", e.0)
            )?;

        return Ok(());
    }

    fn reconnect(&self) -> SyRes<()>
    {
        return
            self
                .stream
                .reconnectlog();
    }

    fn update_tap_data(&self, tap_data: D) -> SyRes<()>
    {
        return
            self
                .stream
                .update_tap_data(tap_data);
    }

}

impl<'stream, F: SyslogFormatter, D: SyslogDestination> SyStreamApi<'stream, F, D, SyncSyslog<F, D>> 
for SyncSyslog<F, D>
{
    fn stream(&'stream self, pri: Priority) -> SyStream<'stream, D, F, SyncSyslog<F, D>> 
    {
        return 
            SyStream 
            { 
                inner: self, 
                pri: pri, 
                _p: PhantomData, 
                _p1: PhantomData 
            };
    }
}

#[cfg(test)]
mod tests
{
    use crate::
    {
       error::SyslogErrCode, LogFacility, LogStat, Priority, SyStreamApi, SyncSyslog, SyslogApi, SyslogLocal 
    };

    #[test]
    fn test_single_message()
    {

        let log = 
            SyncSyslog::openlog(
                Some("test1"), 
                LogStat::LOG_CONS | LogStat::LOG_NDELAY | LogStat::LOG_PID, 
                LogFacility::LOG_DAEMON,
            SyslogLocal::new()
        );

        assert_eq!(log.is_ok(), true, "{}", log.err().unwrap());

        let log = log.unwrap();

        let msg1 = format!("test UTF-8 проверка BOM UTF-8");
        let now = std::time::Instant::now();

        log.syslog(Priority::LOG_DEBUG, msg1.into());

        let dur = now.elapsed();
        println!("{:?}", dur);

        let msg2 = format!("test UTF-8  きるさお命泉ぶねりよ日子金れっ");

        let now = std::time::Instant::now();

        log.syslog(Priority::LOG_DEBUG, msg2.into());

        let dur = now.elapsed();
        println!("{:?}", dur);

        let _ = log.closelog();

        return;
    }

    #[test]
    fn test_single_stream_test()
    {
        use std::fmt::Write;

        let log = 
            SyncSyslog::openlog(
                Some("test1"), 
                LogStat::LOG_CONS | LogStat::LOG_NDELAY | LogStat::LOG_PID, 
                LogFacility::LOG_DAEMON,
            SyslogLocal::new());

        assert_eq!(log.is_ok(), true, "{}", log.err().unwrap());

        let log = log.unwrap();
        
        write!(log.stream(Priority::LOG_DEBUG), "test123").unwrap();

    }

    #[test]
    fn test_single_message_withound_nd()
    {

        let log = 
            SyncSyslog::openlog(
                Some("test1_nond"), 
                LogStat::LOG_CONS | LogStat::LOG_PID, 
                LogFacility::LOG_DAEMON,
            SyslogLocal::new());

        assert_eq!(log.is_ok(), true, "{}", log.err().unwrap());

        let log = log.unwrap();

        let msg1 = format!("test UTF-8 проверка BOM UTF-8");
        let now = std::time::Instant::now();

        log.syslog(Priority::LOG_DEBUG, msg1.into());

        let dur = now.elapsed();
        println!("{:?}", dur);

        let msg2 = format!("test UTF-8  きるさお命泉ぶねりよ日子金れっ");

        let now = std::time::Instant::now();

        log.syslog(Priority::LOG_DEBUG, msg2.into());

        let dur = now.elapsed();
        println!("{:?}", dur);

        let _ = log.closelog();

        return;
    }

    #[test]
    fn test_single_message_perror()
    {
        /*use std::sync::Arc;
        use std::thread;
        use std::time::Duration;
        use super::{LOG_MASK};*/

        let log = 
            SyncSyslog::openlog(
                    Some("test2"), 
                    LogStat::LOG_CONS | LogStat::LOG_NDELAY | LogStat::LOG_PID | LogStat::LOG_PERROR, 
                    LogFacility::LOG_DAEMON,
                    SyslogLocal::new()
                );

        assert_eq!(log.is_ok(), true, "{}", log.err().unwrap());

        let log = log.unwrap();

        log.syslog(Priority::LOG_DEBUG, format!("perror test UTF-8 きるさお命泉ぶねりよ日子金れっ проверка BOM").into());

        let _ = log.closelog();

        return;
    }

    #[test]
    fn test_multithreading()
    {
        use std::sync::Arc;
        use std::thread;
        use std::time::{Instant, Duration};

        let log = 
                SyncSyslog::openlog(
                    Some("test3"), 
                    LogStat::LOG_CONS | LogStat::LOG_NDELAY | LogStat::LOG_PID, 
                    LogFacility::LOG_DAEMON,
                SyslogLocal::new()
            );

        assert_eq!(log.is_ok(), true, "{}", log.err().unwrap());

        let log = Arc::new(log.unwrap());
        let c1_log = log.clone();
        let c2_log = log.clone();

        thread::spawn(move|| {
            for i in 0..5
            {
                //thread::sleep(Duration::from_nanos(200));
                let fmmt = format!("a message from thread 1 #{}[]", i);
                let now = Instant::now();
                c1_log.syslog(Priority::LOG_DEBUG, fmmt.into());
                let elapsed = now.elapsed();
                println!("t1: {:?}", elapsed);
            }
        });

        thread::spawn(move|| {
            for i in 0..5
            {
                //thread::sleep(Duration::from_nanos(201));
                let fmmt = format!("きるさお命泉ぶねりよ日子金れっ {}", i);
                let now = Instant::now();
                c2_log.syslog(Priority::LOG_DEBUG, fmmt.into());
                let elapsed = now.elapsed();
                println!("t2: {:?}", elapsed);
            }
        });

        let now = Instant::now();
        log.syslog(Priority::LOG_DEBUG, format!("A message from main, きるさお命泉ぶねりよ日子金れっ").into());
        let elapsed = now.elapsed();
        println!("main: {:?}", elapsed);

        thread::sleep(Duration::from_secs(2));

        let _ = log.closelog();

        return;
    }

    #[test]
    fn test_multithreading_rw()
    {
        use std::sync::Arc;
        use std::thread;
        use std::time::{Instant, Duration};

        let log = 
                SyncSyslog::openlog(
                    Some("test3"), 
                    LogStat::LOG_CONS | LogStat::LOG_NDELAY | LogStat::LOG_PID, 
                    LogFacility::LOG_DAEMON,
                SyslogLocal::new()
            );

        assert_eq!(log.is_ok(), true, "{}", log.err().unwrap());

        let log = Arc::new(log.unwrap());
        let c1_log = log.clone();

        let thread_handler = 
            thread::spawn(move|| 
                {
                    std::thread::park();
                    for i in 0..30
                    {
                        

                        let fmmt = format!("a message from thread 1 #{}[]", i);
                        let now = Instant::now();
                        c1_log.syslog(Priority::LOG_DEBUG, fmmt.into());
                        let elapsed = now.elapsed();
                        println!("t1: {:?}", elapsed);
                    }
                }
            );

        let now = Instant::now();
        log.syslog(Priority::LOG_DEBUG, format!("A message from main, きるさお命泉ぶねりよ日子金れっ").into());
        let elapsed = now.elapsed();
        println!("main: {:?}", elapsed);

        thread_handler.thread().unpark();

        for i in 0..3
        {
            std::thread::sleep(Duration::from_nanos(999));
            let identity = format!("identity{}", i);
            let now = Instant::now();
            log.change_identity(Some(&identity)).unwrap();
            let elapsed = now.elapsed();
            println!("idc: {:?}", elapsed);
        }

        thread_handler.join().unwrap();

        let _ = log.closelog();

        return;
    }

    #[test]
    fn test_multithreading_rw_sock()
    {
        use std::sync::Arc;
        use std::thread;
        use std::time::{Instant, Duration};

        let log = 
                SyncSyslog::openlog(
                    Some("test3"), 
                    LogStat::LOG_CONS | LogStat::LOG_NDELAY | LogStat::LOG_PID, 
                    LogFacility::LOG_DAEMON,
                SyslogLocal::new()
            );

        assert_eq!(log.is_ok(), true, "{}", log.err().unwrap());

        let log = Arc::new(log.unwrap());
        let c1_log = log.clone();

        let thread_handler = 
            thread::spawn(move|| 
                {
                    std::thread::park();
                    for i in 0..30
                    {
                        

                        let fmmt = format!("a message from thread 1 #{}[]", i);
                        let now = Instant::now();
                        c1_log.syslog(Priority::LOG_DEBUG, fmmt.into());
                        let elapsed = now.elapsed();
                        println!("t1: {:?}", elapsed);
                    }
                }
            );

        let now = Instant::now();
        log.syslog(Priority::LOG_DEBUG, format!("A message from main, きるさお命泉ぶねりよ日子金れっ").into());
        let elapsed = now.elapsed();
        println!("main: {:?}", elapsed);

        thread_handler.thread().unpark();

        for _ in 0..3
        {
            std::thread::sleep(Duration::from_nanos(999));
            let now = Instant::now();
            log.update_tap_data(SyslogLocal::new()).unwrap();
            let elapsed = now.elapsed();
            println!("idc: {:?}", elapsed);
        }

        thread_handler.join().unwrap();

        let _ = log.closelog();

        return;
    }

    #[test]
    fn test_multithreading_simlock()
    {
        use std::sync::Arc;
        use std::thread;
        use std::time::Instant;

        let log = 
                SyncSyslog::openlog(
                    Some("test3"), 
                    LogStat::LOG_CONS | LogStat::LOG_NDELAY | LogStat::LOG_PID, 
                    LogFacility::LOG_DAEMON,
                SyslogLocal::new()
            );

        assert_eq!(log.is_ok(), true, "{}", log.err().unwrap());

        let log = Arc::new(log.unwrap());
        for _ in 0..5
        {
            let c1_log = log.clone();

            let thread_handler = 
                thread::spawn(move || 
                    {
                        std::thread::park();
                        let res = c1_log.update_tap_data(SyslogLocal::new());

                        return res;
                    }
                );

            let now = Instant::now();
            log.syslog(Priority::LOG_DEBUG, format!("A message from main, きるさお命泉ぶねりよ日子金れっ").into());
            let elapsed = now.elapsed();
            println!("main: {:?}", elapsed);

            thread_handler.thread().unpark();

            let res = log.update_tap_data(SyslogLocal::new());

            let res2 = thread_handler.join().unwrap();

            if res.is_ok() == true && res2.is_ok() == true
            {
                continue;
            }
            else if res.is_err() == true
            {
                println!("res err: {}", res.as_ref().err().unwrap());
                assert_eq!(res.err().unwrap().get_errcode(), SyslogErrCode::CoWAlreadyLocked);
            }
            else if res2.is_err() == true
            {
                println!("res2 err: {}", res2.as_ref().err().unwrap());
                assert_eq!(res2.err().unwrap().get_errcode(), SyslogErrCode::CoWAlreadyLocked);
            }

            return;
        }

        panic!("faild!");
    }

    #[test]
    fn long_msg_test()
    {
        // 40 EMSGSIZE
        let msg = 
    "7赤ざクぽな載覧な改申ほふ取容う主坊酸ヱ司戦ヒソオ力端めゃ間真ル実記キ団需くスルて学回県仁京ぴ熱完かあもく。上ょぜ催5強変却ソヲキ転入ク記事購シリ断衝ぽ玲面たぽつへ様態の善無かー勢加ヨマナキ趣会撮さはこ違42器のレうゆ。稿エスタ己報照アイネル岩撲ムニ半者サアキツ画47込死請誌8策ノ再然ださそ禁断しにた高否リ続最ユケ山芸ロ去群づへ索芭あン掲佳怖斎澤クね。
    書キマコヒ上広ざばてわ会佐のにりリ学総点フわすか頼野ぜイよば株約ネキヱメ必軽チヲ録使ラ下能ナウコ紀捜れ霊別イ摯療じらどト相3聞めびら情拳ユフエ確経板植えーぜあ。無ど曲注程クタル系新どばくい都面村リと計安ワヱ月見モヌエノ完臨健85定テサコミ貸容ミテタカ写載とかえげ詐訃だみそな。代け設養テヨナ乱54服料ニ止画ら不暮ッ強治ぱンやし供方広づじもあ般判ごくドラ示刊ヱサトキ速事取さふぞ授共西れッ。
    陸オク康59面くこ惑7伊エノモ候余ロソウ政投ナク転文だけス香両4誕よ数真ひぞぴざ空加ユラ勢隣ゃびよ移72士シヌ図際想カマテ覧費活輸権因ん。東がイ価変や濃恒ヒメネエ学見ょ理供カオムヱ針中わば文63転ヨリホミ礼民ネトマツ速果還ばが転世晴げでんが。犯レト同岸サケ写新ゆぐぱひ人捕けほゅえ的人第いぜ転連ぎど京分株ニムトヒ家及ユフケヌ記定み歩死質をゃいイ都末ワ革量んつ覧打百レいん。
    下ゃとぞら川株以東ぴそぱ費更ウマヨメ覧論ウスレモ八度ずんとイ発平う果構つ小各ぼス夢国フラ連石はちい雪飲聴ゅいと。門ぱし続消上ト文影投設テユチ設川ルラシケ引押ヤマセメ権書ハテ例発リロソイ環載ぴ企実カマリト店鉄ワソロフ逆子先チルマ楽法どゆね声東女携抜爆んけみ。14報ふラめて怒止に開謙こさ促大ヘチツ遺飯ケ営長コキ価空少ろッひぱ読信ケテウ者育ヒミク代57柱浪フ。
    ";

    println!("{}", msg.len());

        let log = 
                SyncSyslog::openlog(
                    Some("test4"), 
                    LogStat::LOG_CONS | LogStat::LOG_NDELAY | LogStat::LOG_PID, 
                    LogFacility::LOG_DAEMON,
                    SyslogLocal::new()
                );

        assert_eq!(log.is_ok(), true, "{}", log.err().unwrap());

        let log = log.unwrap();

        log.syslog(Priority::LOG_DEBUG, format!("{}", msg).into());
                    
        let _ = log.closelog();

        return;
    }
}