syslog_rs/a_sync/

syslog_async.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::marker::PhantomData;

use crate::
{
    a_sync::
    {
        syslog_async_internal::{AsyncMutexGuard, AsyncSyslogInternal}
    }, 
    error::SyRes, 
    formatters::{SyslogFormatter}, 
    syslog_provider::*, 
    LogFacility, 
    LogStat, 
    Priority
};

use crate::a_sync::syslog_async_internal::AsyncMutex;

#[cfg(feature = "build_async_interface")]
use crate::a_sync::syslog_async_internal::AsyncSyslogInternalIO;

#[cfg(feature = "async_embedded")]
use crate::a_sync::{syslog_async_internal::AsyncSyslogInternalIO, DefaultAsyncMutex};

#[cfg(feature = "async_embedded")]
use crate::formatters::DefaultSyslogFormatter;

#[cfg(feature = "async_embedded")]
use crate::a_sync::DefaultIOs;

use super::{syslog_trait::AsyncSyslogApi};


/// A main instance of the Syslog client.
/// 
/// * `D` - a [SyslogDestination] instance which is either:
///     [SyslogLocal], [SyslogFile], [SyslogNet], [SyslogTls] or other. By
///     default a `SyslogLocal` is selected. 
/// 
/// * 'F' - a [SyslogFormatter] formatter which should format the message for the
///     [SyslogDestination] (`D`). By deafult, the [DefaultSyslogFormatter] is used which
///     automatically selects the formatter which is used by syslog on the current system.
///     If current crate does not have a build-in formatter, then a new one should be created.
/// 
/// * 'IO' - a [AsyncSyslogInternalIO] an additional IO like writing to syscons or stderr. And
///     thread operations like sleep. By default a [DefaultIOs] is used which automatically picks
///     correct instance.
/// 
/// * 'MUX` - a [AsyncMutex] implementation above the mutex instance. By default, a [DefaultAsyncMutex]
///     is used which autoselects the correct mutex implementation.
#[cfg(feature = "async_embedded")]
#[derive(Debug)]
pub struct AsyncSyslog<D = SyslogLocal, F = DefaultSyslogFormatter, IO = DefaultIOs, MUX = DefaultAsyncMutex<F, D, IO>>
    (MUX, PhantomData<D>, PhantomData<F>, PhantomData<IO>)
where 
    D: AsyncSyslogDestination, 
    F: SyslogFormatter + Sync, 
    MUX: AsyncMutex<F, D, AsyncSyslogInternal<F, D, IO>>,
    IO: AsyncSyslogInternalIO;
 
/// A main instance of the Syslog client.
/// 
/// * `D` - a [SyslogDestination] instance which is either:
///     [SyslogLocal], [SyslogFile], [SyslogNet], [SyslogTls] or other. The caller
///     should implement the [SyslogDestination] and pass the instance to the crate.
/// 
/// * 'F' - a [SyslogFormatter] formatter which should format the message for the
///     [SyslogDestination] (`D`). The caller should select the correct syslog formatter
///     from provided or use autotype [DefaultSyslogFormatter] or create own.
/// 
/// * 'IO' - a [AsyncSyslogInternalIO] an additional IO like writing to syscons or stderr. And
///     thread operations like sleep. A caller should implement the [AsyncSyslogInternalIO]
///     based on the async executer is used.
/// 
/// * 'MUX` - a [AsyncMutex] implementation above the mutex instance. The caller should implement 
///     this trait and pass the implementation to the struct.
#[cfg(feature = "build_async_interface")]
#[derive(Debug)]
pub struct AsyncSyslog<D, F, IO, MUX>
    (MUX, PhantomData<D>, PhantomData<F>, PhantomData<IO>)
where 
    D: AsyncSyslogDestination, 
    F: SyslogFormatter + Sync, 
    MUX: AsyncMutex<F, D, AsyncSyslogInternal<F, D, IO>>,
    IO: AsyncSyslogInternalIO;

#[cfg(feature = "async_embedded")]
impl AsyncSyslog
{
    /// Opens a default async connection to the local syslog server with default formatter.
    /// 
    /// # 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` -a [SyslogLocal] instance with configuration.
    /// 
    /// # Returns
    /// 
    /// A [SyRes] is returned ([Result]) with: 
    /// 
    /// * [Result::Ok] - with instance
    /// 
    /// * [Result::Err] - with error description.
    pub async 
    fn openlog(ident: Option<&str>, logstat: LogStat, facility: LogFacility, net_tap: SyslogLocal) -> SyRes<Self>
    {        
         let mut syslog = 
            AsyncSyslogInternal::<DefaultSyslogFormatter, SyslogLocal, DefaultIOs>::new(ident, logstat, facility, net_tap)?;
       
        if logstat.contains(LogStat::LOG_NDELAY) == true
        {
            syslog.connectlog().await?;
        }
        
        let mux_syslog = DefaultAsyncMutex::a_new(syslog);
        
        return Ok( 
            Self(mux_syslog, PhantomData::<SyslogLocal>, PhantomData::<DefaultSyslogFormatter>, PhantomData::<DefaultIOs>) 
        );
    }
}

/// A shared implementation.
impl<F, D, IO, MUX> AsyncSyslog<D, F, IO, MUX>
where 
    F: SyslogFormatter + Sync, 
    D: AsyncSyslogDestination, 
    MUX: AsyncMutex<F, D, AsyncSyslogInternal<F, D, IO>>,
    IO: AsyncSyslogInternalIO
{
    /// Opens a special connection to the destination syslog server with specific formatter.
    /// 
    /// All struct generic should be specified before calling this function.
    /// 
    /// # 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` - 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 async 
    fn openlog_with(ident: Option<&str>, logstat: LogStat, facility: LogFacility, net_tap: D) -> SyRes<AsyncSyslog<D, F, IO, MUX>>
    {
        let mut syslog = 
            AsyncSyslogInternal::<F, D, IO>::new(ident, logstat, facility, net_tap)?;
       
        if logstat.contains(LogStat::LOG_NDELAY) == true
        {
            syslog.connectlog().await?;
        }
        
        let mux_syslog = MUX::a_new(syslog);
        
        return Ok( Self(mux_syslog, PhantomData::<D>, PhantomData::<F>, PhantomData::<IO>) );
    }

    /// Sets the logmask to filter out the syslog calls.
    /// 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) 
    #[inline]
    pub async 
    fn setlogmask(&self, logmask: i32) -> i32
    {           
        return 
            self
                .0
                .a_lock()
                .await
                .guard_mut()
                .set_logmask(logmask);
    }

    /// Changes the identity i.e program name which will appear on the logs.
    /// 
    /// Can return error if mutex is poisoned.
    pub async 
    fn change_identity(&self, ident: &str)
    {
        return 
            self
                .0
                .a_lock()
                .await
                .guard_mut()
                .change_identity(ident);
    }

    /// Closes connection to the syslog server
    pub async 
    fn closelog(&self) -> SyRes<()>
    {
        return 
            self
                .0
                .a_lock()
                .await
                .guard_mut()
                .closelog()
                .await;
    }

    /// Similar to libc, syslog() sends data to syslog server.
    /// 
    /// # Arguments
    ///
    /// * `pri` - a priority [Priority]
    ///
    /// * `fmt` - a program's message to be sent as payload.
    #[inline]
    pub async 
    fn syslog(&self, pri: Priority, fmt: String)
    {
        self.0.a_lock().await.guard_mut().vsyslog1(pri, fmt.into()).await;
    }

    /// Sends message to syslog (same as `syslog`).
    #[inline]
    pub async
    fn vsyslog(&self, pri: Priority, fmt: &'static str)
    {
        self.0.a_lock().await.guard_mut().vsyslog1(pri, fmt.into()).await;
    }

    /// Sends the specificly formatted message i.e RFC5424 allows to send additional data
    /// like STRUCTURED-DATA, SD-ID, SD-PARAM.
    #[inline]
    pub async 
    fn esyslog(&self, pri: Priority, fmt: F)
    {
        self.0.a_lock().await.guard_mut().vsyslog1(pri, fmt).await;
    }

    /// 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
    pub async 
    fn reconnect(&self) -> SyRes<()>
    {
        return self.0.a_lock().await.guard_mut().reconnect().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
    pub async 
    fn update_tap(&self, new_tap: D) -> SyRes<()>
    {
        return self.0.a_lock().await.guard_mut().update_tap_data(new_tap).await;
    }
}



#[cfg(test)]
mod tests
{
   

    use super::*;

    #[cfg(feature = "build_async_smol")]
    #[test]
    fn test_smol() -> smol::io::Result<()> 
    {
        smol::block_on(
                async 
                {
                    use std::{sync::Arc, time::{Duration, Instant}};

                    use smol::Timer;

                    let log =
                        AsyncSyslog::openlog(
                                Some("smol_test1"), 
                                LogStat::LOG_CONS | LogStat::LOG_NDELAY | LogStat::LOG_PID, 
                                LogFacility::LOG_DAEMON,
                                SyslogLocal::new()
                            )
                            .await;

                    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();

                    smol::spawn(async move 
                        {
                            for i in 0..5
                            {
                                use std::time::Duration;

                                use smol::Timer;

                                let cc_c1_log = c1_log.clone();
                                Timer::after(Duration::from_nanos(200)).await;
                                smol::spawn( async move 
                                {
                                    use std::time::Instant;

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

                    smol::spawn(async move 
                        {
                            for i in 0..5
                            {
                                use std::time::Duration;

                                use smol::Timer;

                                let cc_c2_log = c2_log.clone();
                                Timer::after(Duration::from_nanos(201)).await;
                                smol::spawn( async move 
                                {
                                    use std::time::Instant;

                                    let m = format!("ASYNC きるさお命泉ぶねりよ日子金れっ {}", i);
                                    let now = Instant::now();
                                    cc_c2_log.syslog(Priority::LOG_DEBUG, m.into()).await;
                                    let elapsed = now.elapsed();
                                    println!("t2: {:?}", elapsed);
                                }).await;
                            }
                        }).await;

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

                     log.change_identity("smol_test1new").await;

                    let m = format!("ASYNC A message from main new ident, きるさお命泉ぶねりよ日子金れっ");

                    log.syslog(Priority::LOG_DEBUG, m).await;

                    Timer::after(Duration::from_secs(1)).await;

                    log.closelog().await.unwrap();

                    Timer::after(Duration::from_nanos(201)).await;

                    Ok(())
                }
            )
    }

    #[cfg(feature = "build_async_tokio")]
    #[tokio::test]
    async fn test_multithreading()
    {
        use tokio::time::Instant;
        use std::sync::Arc;
        use tokio::time::{sleep, Duration};
        
        let log = 
            AsyncSyslog::openlog(
                Some("asynctest1"), 
                LogStat::LOG_CONS | LogStat::LOG_NDELAY | LogStat::LOG_PID, 
                LogFacility::LOG_DAEMON,
                SyslogLocal::new()
            ).await;

        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();

        tokio::spawn( async move 
            {
                for i in 0..5
                {
                    let cc_c1_log = c1_log.clone();
                    sleep(Duration::from_nanos(200)).await;
                    tokio::spawn( async move 
                    {
                        let m = format!("ASYNC a message from thread 1 #{}[]", i);
                        let now = Instant::now();
                        cc_c1_log.syslog(Priority::LOG_DEBUG, m).await;
                        let elapsed = now.elapsed();
                        println!("t1: {:?}", elapsed);
                    });
                }
            }
        );

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

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

        
        sleep(Duration::from_secs(1)).await;

        log.change_identity("asynctest1new").await;

        let m = format!("ASYNC A message from main new ident, きるさお命泉ぶねりよ日子金れっ");

        log.syslog(Priority::LOG_DEBUG, m).await;

        log.closelog().await.unwrap();

        sleep(Duration::from_nanos(201)).await;

        return;
    }
}