/*-
 * syslog-rs - a syslog client translated from libc to rust
 * Copyright (C) 2020  Aleksandr Morozov, RELKOM s.r.o
 * Copyright (C) 2021-2022  Aleksandr Morozov
 * 
 * This Source Code Form is subject to the terms of the Mozilla Public
 * License, v. 2.0. If a copy of the MPL was not distributed with this
 *  file, You can obtain one at https://mozilla.org/MPL/2.0/.
 */


use std::fmt::{Write, self, Arguments};
use std::path::Path;
use std::thread;
use std::sync::{Arc, Weak};
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;

use crossbeam::channel::{unbounded, Sender, Receiver};
use crossbeam::deque::{Injector, Steal};

use crate::{map_error, throw_error_code, map_error_code};
use crate::common::*;
use crate::error::{SyRes, SyslogError, SyslogErrCode};

use super::syslog_stream::SyslogStream;
use super::syslog_sync_internal::SyncSyslogInternal;
pub use super::syslog_trait::{SyslogStd, SyslogExt};

/// A wrappes for the data in the queue
enum SyCmd
{
    /// A message to syslog server
    Syslog
    {
        pri: i32,
        msg: String
    },

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

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

    /// A request to stop processing and quit
    Stop,
}

impl SyCmd
{
    fn form_syslog(pri: i32, msg: String) -> Self
    {
        return Self::Syslog
            {
                pri, msg
            };
    }

    fn form_logmask(logmask: i32, need_prev_pri: bool) -> (Self, Option<Receiver<i32>>)
    {
        let ret =
            if need_prev_pri == true
            {
                let (tx, rx) = unbounded::<i32>();

                (Self::Logmask{logmask, loopback: Some(tx)}, Some(rx))
            }
            else
            {
                (Self::Logmask{logmask, loopback: None}, None)
            };

        return ret;
    }

    fn form_change_ident(identity: String) -> Self
    {
        return Self::ChangeIdentity
            {
                identity: identity
            };
    }

    fn form_stop() -> Self
    {
        return Self::Stop;
    }
}

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

    /// commands channel
    tasks: Arc<Injector<SyCmd>>,

    /// An syslog assets
    inner: SyncSyslogInternal,
}

impl Drop for SyslogInternal
{
    fn drop(&mut self)
    {
        self.inner.disconnectlog();
    }
}

impl SyslogInternal
{
    fn new(
        run_flag: Arc<AtomicBool>, 
        tasks: Arc<Injector<SyCmd>>,
        ident: Option<&str>, 
        logstat: LogStat, 
        facility: LogFacility,
        opt_path: Option<&Path>,
    ) -> SyRes<Self>
    {
        let mut ret = 
            Self
            {
                run_flag: run_flag,
                tasks: tasks,
                inner: SyncSyslogInternal::new(ident, logstat, facility, opt_path)
            };

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

        return Ok(ret);
    }

    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.steal()
			{
				Steal::Success(task) =>
				{
                    match task
                    {
                        SyCmd::Syslog{pri, msg} =>
                        {
                            self.inner.vsyslog1(pri, msg);
                        },
                        SyCmd::Logmask{logmask, loopback} =>
                        {
                            let pri = self.inner.set_logmask(logmask);

                            if let Some(lbk) = loopback
                            {
                                let _ = lbk.send(pri);
                            }
                        },
                        SyCmd::ChangeIdentity{identity} =>
                        {
                            self.inner.set_logtag(identity);
                        },
                        SyCmd::Stop =>
                        {
                            // ignore the rest
                            break;
                        }
                    }
                },
                Steal::Retry =>
				{
                    // retry as advised
                    thread::sleep(Duration::from_nanos(100));
                },
                _ => 
				{
                    thread::park_timeout(Duration::from_millis(500));
                }
            } // match

        } // loop

        return;
    }
}

/// A common instance which describes the syslog state
pub struct Syslog
{   
    /// Control flag
    run_control: Weak<AtomicBool>,

    /// commands channel
    tasks: Arc<Injector<SyCmd>>,

    /// process thread
    thread: Option<thread::JoinHandle<()>>,
}

unsafe impl Send for Syslog {}
unsafe impl Sync for Syslog {}


impl SyslogStd for Syslog
{
    /// As in a libc, this function initializes the syslog instance. The 
    /// main difference with realization in C is it returns the instance
    /// to program used this crate. This structure implements the [Send] 
    /// and [Sync] so it does not require any additional synchonization.
    ///
    /// # Arguments
    /// 
    /// * `ident` - a identification of the sender. If not set, the crate
    ///             will determine automatically!
    /// * `logstat` - sets up the syslog behaviour. Use [LogStat] 
    /// 
    /// * `facility` - a syslog facility. Use [LogFacility]
    /// 
    /// # Returns
    ///
    /// * A [SyRes] with instance or Err()
    ///
    /// # Example
    /// 
    /// ```
    ///  Syslog::openlog(
    ///        Some("test1"), 
    ///         LogStat::LOG_NDELAY | LogStat::LOG_PID, 
    ///         LogFacility::LOG_DAEMON);
    /// ``` 
    fn openlog(ident: Option<&str>, logstat: LogStat, facility: LogFacility) -> SyRes<Self>
    {
        // control flag
        let run_flag: Arc<AtomicBool> = Arc::new(AtomicBool::new(true));
        let run_control = Arc::downgrade(&run_flag);

        // creating queue for messages
		let tasks = 
            Arc::new(Injector::<SyCmd>::new());

        // creating internal syslog struct
        let inst = 
            SyslogInternal::new(run_flag, tasks.clone(), ident, logstat, facility, None)?;

        // initiate a thread
        let thread_hnd = 
            thread::Builder::new()
                .name("syslog/0".to_string())
                .spawn(move || SyslogInternal::thread_worker(inst))
                .map_err(|e| 
                    map_error!("ctor Parser: thread spawn failed. {}", e)
                )?;
        
        // creating a syslog public struct instance
        let ret = 
            Self
            {
                run_control: run_control,
                tasks: tasks,
                thread: Some(thread_hnd),
            };

        return Ok(ret);
    }

    /// 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) 
    fn setlogmask(&self, logmask: i32) -> SyRes<i32>
    {
        if self.run_control.upgrade().is_some() == true
        {
            let (sy_cmd, opt_rx) = 
                SyCmd::form_logmask(logmask, true);

            self.tasks.push(sy_cmd);

            self.thread.as_ref().unwrap().thread().unpark();

            let rx = opt_rx.unwrap();

            return 
                rx.recv()
                    .map_err(|e| 
                        map_error_code!(SyslogErrCode::UnboundedChannelError, "{}", e)
                    );
        }

        throw_error_code!(SyslogErrCode::SyslogThreadNotAvailable, "syslog is not available");
    }

    /// Similar to libc, closelog() will close the log 
    fn closelog(&self) -> SyRes<()>
    {
        if self.run_control.upgrade().is_some() == true
        {
            // send stop
            self.tasks.push(SyCmd::form_stop());

            self.thread
                .as_ref()
                .unwrap()
                .thread()
                .unpark();

            return Ok(());
        }

        throw_error_code!(SyslogErrCode::SyslogThreadNotAvailable, "thread is not running");
    }

    /// Similar to libc, syslog() sends data to syslog server.
    /// 
    /// # Arguments
    ///
    /// * `pri` - a priority [Priority]
    ///
    /// * `fmt` - a string message. In C exists a functions with
    ///     variable argumets amount. In Rust you should create your
    ///     own macros like format!() or use format!() 
    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.
        if self.run_control.upgrade().is_some() == true
        {
            let sy_cmd = 
                SyCmd::form_syslog(pri.bits(), fmt);

            self.tasks.push(sy_cmd);

            self.thread.as_ref().unwrap().thread().unpark();
        }

        return;
    }

    /// Similar to syslog() and created for the compatability. 
    fn vsyslog<S: AsRef<str>>(&self, pri: Priority, fmt: S)
    {
        if self.run_control.upgrade().is_some() == true
        {
            let sy_cmd = 
                SyCmd::form_syslog(pri.bits(), fmt.as_ref().to_string());

            self.tasks.push(sy_cmd);

            self.thread.as_ref().unwrap().thread().unpark();
        }

        return;
    }
}

// --- NON STANDART API
impl SyslogExt for Syslog
{
    /// NON STANDARD FUNCTION
    /// 
    /// This function acting like `openlog()` but allows to open connection to 
    /// the arbitrary object.
    /// 
    /// # Arguments
    /// 
    /// * @see `openlog()`
    /// 
    /// * `sock_path` - [AsRef] [Path] a path to the unix datagram socket 
    fn openlog_custom<P>(ident: Option<&str>, logstat: LogStat, facility: LogFacility, sock_path: P) -> SyRes<Self>
    where P: AsRef<Path>
    {
        // control flag
        let run_flag: Arc<AtomicBool> = Arc::new(AtomicBool::new(true));
        let run_control = Arc::downgrade(&run_flag);

        // creating queue for messages
        let tasks = 
            Arc::new(Injector::<SyCmd>::new());

        // creating internal syslog struct
        let inst = 
            SyslogInternal::new(run_flag, tasks.clone(), ident, logstat, facility, Some(sock_path.as_ref()))?;

        // initiate a thread
        let thread_hnd = 
            thread::Builder::new()
                .name("syslog/0".to_string())
                .spawn(move || SyslogInternal::thread_worker(inst))
                .map_err(|e| 
                    map_error!("ctor Parser: thread spawn failed. {}", e)
                )?;

        // creating a syslog public struct instance
        let ret = 
            Self
            {
                run_control: run_control,
                tasks: tasks,
                thread: Some(thread_hnd),
            };

        return Ok(ret);
    }

    /// 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<I: AsRef<str>>(&self, ident: I) -> SyRes<()>
    {
        if self.run_control.upgrade().is_some() == true
        {
            let sy_cmd = 
                SyCmd::form_change_ident(ident.as_ref().to_string());

            self.tasks.push(sy_cmd);

            self.thread.as_ref().unwrap().thread().unpark();
        }

        return Ok(());
    }

    /// Creates an instance which is implements [core::fmt::Write].
    /// 
    /// # Arguments
    /// 
    /// * `pri` - a priority [Priority] (can be updated)
    /// 
    /// # Returns
    /// 
    /// Error not supported
    fn make_stream(&self, pri: Priority) -> Box<dyn SyslogStream>
    {
        return Box::new(
            StreamableSyslog
            {
                inner: self.tasks.clone(),
                pri: pri
            }
        ); 
    }

}

/// An implementation which allows to [write!] to syslog.
/// 
/// Will block awaiting while inner mutex is locked.
/// You need to make sure that [SyslogStream] is stored as
/// mutable instance.
struct StreamableSyslog
{
    /// Syslog instance
    inner: Arc<Injector<SyCmd>>,

    /// Priority
    pri: Priority
}

impl SyslogStream for StreamableSyslog
{
    /// Updates the pri [Priority] 
    fn update_pri(&mut self, new_pri: Priority) -> Priority
    {
        let prev = self.pri;
        self.pri = new_pri;

        return prev;
    }
}

impl Write for StreamableSyslog
{
    fn write_str(&mut self, s: &str) -> fmt::Result 
    {
        let sy_cmd = 
            SyCmd::form_syslog(self.pri.bits(), s.to_string());

        self.inner.push(sy_cmd);

        return Ok(());
    }

    fn write_fmt(self: &mut Self, args: Arguments<'_>) -> fmt::Result
    {
        if let Some(s) = args.as_str() 
        {
            return self.write_str(s);
        } 
        else 
        {
            return self.write_str(&args.to_string());
        }
    }
}


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

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

    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 now = Instant::now();
            c1_log.syslog(Priority::LOG_DEBUG, format!("a message from thread 1 #{}[]", i));
            let elapsed = now.elapsed();
            println!("t1: {:?}", elapsed);
        }
    });

    thread::spawn(move|| {
        for i in 0..5
        {
            thread::sleep(Duration::from_nanos(201));
            let now = Instant::now();
            c2_log.syslog(Priority::LOG_DEBUG, format!("сообщение от треда 2 №{}ХЪ", i));
            let elapsed = now.elapsed();
            println!("t2: {:?}", elapsed);
        }
    });

    let res = log.setlogmask(!LOG_MASK!(Priority::LOG_ERR));

    assert_eq!(res.is_ok(), true, "{}", res.err().unwrap());
    assert_eq!(res.unwrap(), 0xff, "should be 0xff");

    let now = Instant::now();
    log.syslog(Priority::LOG_DEBUG, format!("A message from main, сообщение от главнюка"));
    let elapsed = now.elapsed();
    println!("main: {:?}", elapsed);

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

    let _ = log.closelog();

    thread::sleep(Duration::from_millis(500));

    let res = log.setlogmask(!LOG_MASK!(Priority::LOG_ERR));

    assert_eq!(res.is_err(), true, "not an error, why?");
    let error = res.err().unwrap();
    assert_eq!(error.get_errcode(), SyslogErrCode::SyslogThreadNotAvailable, "unexpected error code {:?}", error);

    return;
}