syslog_rs/a_sync/

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

use crate::{error::SyRes, formatters::{DefaultSyslogFormatter, SyslogFormatter}, syslog_provider::*, LogFacility, LogStat, Priority};

#[cfg(feature = "use_sync")]
use crate::sy_sync::Syslog;


#[cfg(feature = "use_sync_queue")]
use crate::sy_sync_queue::SyslogQueue;
#[cfg(feature = "use_sync_queue")]
use super::syslog_async_queue::AsyncSyslogQueue;

use super::{syslog_async_shared::AsyncSyslogShared, syslog_trait::AsyncSyslogApi};


/// A main instance of the Syslog client.
/// 
/// * `D` - a [SyslogDestination] instance which is either:
///     [SyslogLocal], [SyslogFile], [SyslogNet], [SyslogTls]. By
///     default a `SyslogLocal` is set.
/// 
/// * `F` - a [SyslogFormatter] a message formatter for the destination
///     server. See `formatters` i.e `FormatRfc3146`, `FormatRfc5424`, `FormatFile`. Or
///     `DefaultSyslogFormatter`, `DefaultSyslogFormatterFile`.
/// 
/// * `S` - a [AsyncSyslogApi] sybsystem. It is either [AsyncSyslogShared] or [AsyncSyslogQueue] with 
///     the same 'F' - formatter type. By default a [AsyncSyslogShared] is set.
#[derive(Debug)]
pub struct AsyncSyslog<D = SyslogLocal, F = DefaultSyslogFormatter, S = AsyncSyslogShared<F>>
    (S, PhantomData<F>, PhantomData<D>)
where D: SyslogDestination, F: SyslogFormatter + Sync, S: AsyncSyslogApi;

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.
    /// 
    /// * `dest` - a destination server. Allows to connect to the different local syslog server or
    ///     will be determined automatically.
    /// 
    /// # 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, dest: SyslogLocal) -> SyRes<Self>
    {
        let net_tap = dest.init()?;
        
        let syslog = AsyncSyslogShared::<DefaultSyslogFormatter>::openlog(ident, logstat, facility, net_tap).await?;

        return Ok(Self(syslog, PhantomData::<DefaultSyslogFormatter>, PhantomData::<SyslogLocal>));
    }
}

impl<D: SyslogDestination, F: SyslogFormatter + Sync> AsyncSyslog<D, F, AsyncSyslogShared<F>>
{
    /// 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.
    /// 
    /// * `dest` - a destination server. Allows to connect to the different local syslog server or
    ///     will be determined automatically.
    /// 
    /// # 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, dest: D) -> SyRes<AsyncSyslog<D, F, AsyncSyslogShared<F>>>
    {
        let net_tap = dest.init()?;
        
        let syslog = AsyncSyslogShared::<F>::openlog(ident, logstat, facility, net_tap).await?;
       

        return Ok(Self(syslog, PhantomData::<F>, PhantomData::<D>));
    }

    /// 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) 
    pub async 
    fn setlogmask(&self, logmask: i32) -> SyRes<i32>
    {
        return self.0.setlogmask(logmask).await;
    }

    /// 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) -> SyRes<()>
    {
        return self.0.change_identity(ident).await;
    }

    /// Closes connection to the syslog server
    pub async 
    fn closelog(&self) -> SyRes<()>
    {
        return self.0.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.syslog(pri, fmt).await;
    }

    /// Sends message to syslog (same as `syslog`).
    #[inline]
    pub async
    fn vsyslog<M: AsRef<str>>(&self, pri: Priority, fmt: M)
    {
        self.0.vsyslog(pri, fmt.as_ref()).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.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.update_tap_data(new_tap.init()?).await;
    }
}

#[cfg(feature = "use_sync_queue")]
impl<D: SyslogDestination, F: SyslogFormatter + Sync> AsyncSyslog<D, F, AsyncSyslogQueue<F>>
{
    /// Attaches to the [SyslogQueue] so the async instance of syslog will write to the
    /// same instance with sync portion.
    /// 
    /// # Arguments
    /// 
    /// * `dest` - a [Syslog] instance of type [SyslogQueue]. The `D` [SyslogDestination] of 
    ///     instance does shoud be of the same type as in `sync protion`. 
    /// 
    /// # Returns
    /// 
    /// A [SyRes] is returned. The [Result::Err] will be returned only of thread which serves the
    /// message queue panics or was not started.
    pub async 
    fn attach_queue(dest: &Syslog<D, F, SyslogQueue<F>>) -> SyRes<AsyncSyslog<D, F, AsyncSyslogQueue<F>>>
    {
        let syslog = 
            AsyncSyslogQueue::<F>::attachlog(dest.make_adapter()).await?;
    
        return Ok(Self(syslog, PhantomData::<F>, PhantomData::<D>));
    }

    /// 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) 
    pub async 
    fn setlogmask(&self, logmask: i32) -> SyRes<i32>
    {
        return self.0.setlogmask(logmask).await;
    }
    

    /// Closes connection to the syslog server
    pub async 
    fn closelog(&self) -> SyRes<()>
    {
        return self.0.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.syslog(pri, fmt).await;
    }

    /// Sends message to syslog (same as `syslog`).
    #[inline]
    pub async 
    fn vsyslog<M: AsRef<str>>(&self, pri: Priority, fmt: M)
    {
        self.0.vsyslog(pri, fmt.as_ref()).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.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.update_tap_data(new_tap.init()?).await;
    }
}


#[cfg(test)]
mod tests
{
    use tokio::time::Instant;

    use super::*;

    //#[tokio::test(flavor = "single_thread", worker_threads = 1)]
    #[tokio::test]
    async fn test_multithreading()
    {
        use std::sync::Arc;
        use tokio::time::{sleep, Duration};
        
        let log = 
            AsyncSyslog::openlog(
                Some("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();

        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.vsyslog(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(2)).await;

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

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

        return;
    }
}