Skip to main content

syslog_rs/sync/
syslog_sync_queue.rs

1/*-
2 * syslog-rs - a syslog client translated from libc to rust
3 * 
4 * Copyright 2025 Aleksandr Morozov
5 * 
6 * The syslog-rs crate can be redistributed and/or modified
7 * under the terms of either of the following licenses:
8 *
9 *   1. the Mozilla Public License Version 2.0 (the “MPL”) OR
10 *
11 *   2. The MIT License (MIT)
12 *                     
13 *   3. EUROPEAN UNION PUBLIC LICENCE v. 1.2 EUPL © the European Union 2007, 2016
14 */
15
16
17use std::{fmt};
18use std::sync::{Arc, Weak};
19use std::sync::atomic::{AtomicBool, Ordering};
20
21use crate::formatters::{SyslogFormatter};
22use crate::sync::syslog_sync_internal::SyslogSocketLockless;
23use crate::sync::LogItems;
24
25use crate::{SyslogDestination};
26
27
28use crate::common::*;
29use crate::error::SyRes;
30
31
32/// A wrapper for the data commands in the queue
33pub enum SyCmd<F: SyslogFormatter, D: SyslogDestination, S: SyslogQueueChannel<F, D>>
34{
35    /// A message to syslog server
36    Syslog
37    {
38        pri: Priority,
39        msg: F
40    },
41
42    /// A reuest to change logmask
43    Logmask
44    {
45        logmask: i32, 
46        loopback: S::OneShotChannelSnd<i32>,
47    },
48
49    /// A request to change identity
50    ChangeIdentity
51    {
52        identity: Option<String>,
53    },
54
55    /// Updates the tap settings
56    UpdateTap
57    {
58        tap_type: D,//TapTypeData,
59        loopback: S::OneShotChannelSnd<SyRes<()>>,
60    },
61
62    ConnectLog
63    {
64        loopback: S::OneShotChannelSnd<SyRes<()>>
65    },
66
67    DisconnectLog
68    {
69        loopback: S::OneShotChannelSnd<SyRes<()>>
70    },
71
72    /// A request to rotate file or reconnect.
73    Reconnect,
74
75    /// A request to stop processing and quit
76    #[allow(unused)]
77    Stop,
78}
79
80
81impl<F: SyslogFormatter, D: SyslogDestination, S: SyslogQueueChannel<F, D>> SyCmd<F, D, S>
82{
83    /// Construct a message with data to send.
84    pub(crate) 
85    fn form_syslog(pri: Priority, msg: F) -> Self
86    {
87        return 
88            Self::Syslog
89            {
90                pri, msg
91            };
92    }
93
94    pub(crate) 
95    fn form_connectlog() -> (Self, S::OneShotChannelRcv<SyRes<()>>)
96    {
97        let (tx, rx) = S::create_oneshot_channel::<SyRes<()>>();
98
99        return 
100            (Self::ConnectLog{ loopback: tx }, rx);
101    }
102
103    pub(crate) 
104    fn form_disconnectlog() -> (Self, S::OneShotChannelRcv<SyRes<()>>)
105    {
106        let (tx, rx) = S::create_oneshot_channel::<SyRes<()>>();
107
108        return 
109            (Self::DisconnectLog{ loopback: tx }, rx);
110    }
111
112    /// Constructs a message to make logmask with or without previous PRI.
113    /// 
114    /// # Arguments
115    /// 
116    /// * `logmask` - a new logmask
117    pub(crate) 
118    fn form_logmask(logmask: i32) -> (Self, S::OneShotChannelRcv<i32>)
119    {
120        let (tx, rx) = S::create_oneshot_channel::<i32>();
121
122        return 
123            (Self::Logmask{ logmask, loopback: tx }, rx);
124    }
125
126    /// Constructs a message which should change the identity (appname) of the
127    /// instance.
128    pub(crate) 
129    fn form_change_ident(identity: Option<String>) -> Self
130    {
131        return 
132            Self::ChangeIdentity
133            {
134                identity: identity
135            };
136    }
137
138    /// Constructs a message which changes the destination of the log messages i.e
139    /// changing path of the dst file or address. The `new_tap_type` 
140    /// should be the same variant [TapTypeData] as previous.
141    pub(crate)
142    fn form_update_tap(new_tap_type: D/*TapTypeData*/) -> (Self, S::OneShotChannelRcv<SyRes<()>>)
143    {
144        let (tx, rx) = S::create_oneshot_channel::<SyRes<()>>();
145
146        return (
147            Self::UpdateTap  
148            { 
149                tap_type: new_tap_type,
150                loopback: tx
151            },
152            rx
153        );
154    }
155
156    /// Constructs a message to handle SIGHUP. This is usefull only when the instance
157    /// is writing directly into the file. Or just reconnect.
158    pub(crate) 
159    fn form_reconnect() -> Self
160    {
161        return Self::Reconnect;
162    }
163
164    /// Constructs a message to stop thread gracefully. After receiving this
165    /// message a thread will quit and all messages that would be sent after
166    /// this message will be cleared from queue and a new messages will not be
167    /// received.
168    #[allow(unused)]
169    pub(crate) 
170    fn form_stop() -> Self
171    {
172        return Self::Stop;
173    }
174}
175
176/// Internal struct of the syslog client thread.
177pub(crate) struct SyslogInternal<F: SyslogFormatter, D: SyslogDestination, S: SyslogQueueChannel<F, D>>
178{
179    /// A explicit stop flag
180    run_flag: Arc<AtomicBool>,
181
182    /// commands channel
183    tasks: S::ChannelRcv,
184
185    /// Log config
186    log_items: LogItems,
187
188    /// socket
189    socket: SyslogSocketLockless<D>,
190}
191
192
193
194impl<F, D, S> SyslogInternal<F, D, S>
195where F: SyslogFormatter, D: SyslogDestination, S: SyslogQueueChannel<F, D>
196{
197    pub(crate) 
198    fn new(log_items: LogItems, socket: SyslogSocketLockless<D>) -> SyRes<(Self, S::ChannelSnd, Weak<AtomicBool>)>
199    {
200        // control flag
201        let run_flag: Arc<AtomicBool> = Arc::new(AtomicBool::new(true));
202        let run_control = Arc::downgrade(&run_flag);
203
204        // creating queue for messages
205        let (sender, receiver) = S::create_channel();
206
207        // creating internal syslog struct
208        let mut inst = 
209            SyslogInternal
210            {
211                run_flag: run_flag,
212                tasks: receiver,
213                log_items: log_items,
214                socket: socket
215            };
216
217        if inst.log_items.logstat.contains(LogStat::LOG_NDELAY) == true
218        {
219            inst.socket.connectlog()?;
220        }
221
222        return Ok((inst, sender, run_control));
223    }
224
225    pub(crate) 
226    fn thread_worker(mut self)
227    {
228        loop
229        {
230            // self will be dropped as soon as thread will be stopped
231            if self.run_flag.load(Ordering::Relaxed) == false
232            {
233                // force leave
234                break;
235            }	
236
237            match self.tasks.q_recv_blocking()
238			{
239				Some(task) =>
240				{
241                    match task
242                    {
243                        SyCmd::Syslog{ pri, msg } =>
244                        {
245                            let Some(formatted) = self.log_items.vsyslog1_msg::<F, D>(pri, &msg)
246                                else { continue };
247
248                            let _ = self.socket.vsyslog1(formatted.1, formatted.0);
249                        },
250                        SyCmd::Logmask{ logmask, loopback } =>
251                        {
252                            let pri = self.log_items.set_logmask(logmask);
253
254                            let _ = loopback.send_once_blocking(pri);
255                        },
256                        SyCmd::ChangeIdentity{ identity } =>
257                        {
258                            self.log_items.set_identity(identity.as_ref().map(|v| v.as_str()));
259                        },
260
261                        SyCmd::UpdateTap{ tap_type, loopback } =>
262                        {
263                            let res = self.socket.update_tap_data(tap_type);
264                            
265                            if let Err(Err(e)) = loopback.send_once_blocking(res)
266                            {
267                                self.log_items.logstat.send_to_stderr(&e.to_string());
268                            }
269                        },
270
271                        SyCmd::ConnectLog{ loopback} => 
272                        {
273                            if let Err(Err(e)) = loopback.send_once_blocking(self.socket.connectlog())
274                            {
275                                self.log_items.logstat.send_to_stderr(&e.to_string());
276                            }
277                        },
278
279                        SyCmd::DisconnectLog{ loopback} => 
280                        {
281                            if let Err(Err(e)) = loopback.send_once_blocking(self.socket.disconnectlog())
282                            {
283                                self.log_items.logstat.send_to_stderr(&e.to_string());
284                            }
285                        },
286
287                        SyCmd::Reconnect =>
288                        {
289                            if let Err(e) = self.socket.disconnectlog()
290                            {
291                                self.log_items.logstat.send_to_stderr(&e.to_string());
292                            }
293                                
294                            if let Err(e) = self.socket.connectlog()
295                            {
296                                self.log_items.logstat.send_to_stderr(&e.to_string());
297                            }
298                        },
299                        SyCmd::Stop =>
300                        {
301                            // ignore the rest
302                            break;
303                        }
304                    }
305                },
306                None =>
307                {
308                    break;
309                }
310            } // match
311
312        } // loop
313
314        return;
315    }
316}
317
318/// A trait which should be implemented by the channel provider which forms a command queue.
319/// This trait provides a blocking receive interface on the receiver side.
320pub trait SyslogQueueChanRcv<F: SyslogFormatter, D: SyslogDestination, S: SyslogQueueChannel<F, D>>: fmt::Debug + Send
321{
322    /// Receive from channel in blocking mode.
323    fn q_recv_blocking(&mut self) -> Option<SyCmd<F, D, S>>;
324}
325
326/// A trait which should be implemented by the channel provider which forms a command queue.
327/// This trait provides both or either the blocking send interface.
328#[allow(async_fn_in_trait)]
329pub trait SyslogQueueChanSnd<F: SyslogFormatter, D: SyslogDestination, S: SyslogQueueChannel<F, D>>: fmt::Debug + Clone
330{
331    /// Sends the message over the channel in blocking mode.
332    fn q_send_blocking(&self, msg: SyCmd<F, D, S>) -> SyRes<()>;
333
334    /// Sends the message over the channel in async mode. By default, it returns error.
335    async fn q_send(&self, _msg: SyCmd<F, D, S>) -> SyRes<()>
336    {
337        crate::throw_error!("async is not availabe here"); 
338    }
339}
340
341/// A trait which should be implemented by the channel provider which forms a command queue.
342/// This trait provides a Oneshot channel a receive interface for both sync and async modes.
343#[allow(async_fn_in_trait)]
344pub trait SyslogQueueOneChanRcv<C>
345{
346    /// Receives once from the channel in sync mode.
347    fn recv_once_blocking(self) -> SyRes<C>;
348
349    /// Receives once from the channel in async mode. By default, it returns error if not 
350    /// implemented.
351    async fn recv_once(self) -> SyRes<C> where Self: Sized
352    {
353        crate::throw_error!("async is not availabe here"); 
354    }
355}
356
357/// A trait which should be implemented by the channel provider which forms a command queue.
358/// This trait provides a Oneshot channel a send interface for blocking mode.
359pub trait SyslogQueueOneChanSnd<C: Send>: fmt::Debug + Send
360{
361    /// Send to channel in blocing mode.
362    fn send_once_blocking(self, data: C) -> Result<(), C>;
363}
364
365/// A trait which should be implemented by the channel provider which forms a command queue.
366/// This trait provides a common interface which includes everything i.e channel, oneshot 
367/// channel and manipulations.
368pub trait SyslogQueueChannel<F: SyslogFormatter, D: SyslogDestination>: fmt::Debug + Send + Clone + 'static
369{
370    const ADAPTER_NAME: &'static str;
371    
372    /// A send side of the channel type.
373    type ChannelSnd: SyslogQueueChanSnd<F, D, Self> + Sync;
374    
375    /// A receive side of the channel type.
376    type ChannelRcv: SyslogQueueChanRcv<F, D, Self>;
377
378    /// A oneshot send side of the channel type.
379    type OneShotChannelSnd<C: Send + fmt::Debug>: SyslogQueueOneChanSnd<C>;
380
381    /// A oneshor receive side of the channel type.
382    type OneShotChannelRcv<C>: SyslogQueueOneChanRcv<C>;
383
384    /// Creates unbounded channel.
385    fn create_channel() -> (Self::ChannelSnd, Self::ChannelRcv);   
386
387    /// Creates oneshot channel.
388    fn create_oneshot_channel<C: Send + fmt::Debug>() -> (Self::OneShotChannelSnd<C>, Self::OneShotChannelRcv<C>);
389}
390
391/*
392
393/// A parallel, shared instance of the syslog client which is running in the 
394/// separate thread and uses a crossbeam channel to receive the messages from 
395/// the program. It is also capable to combine sync and async i.e sync code and
396/// async code is writing to the same syslog connection.
397/// 
398/// For this isntance a [SyslogApi] and [SyStreamApi] are implemented.
399/// 
400/// Also if `async` is enabled, a [AsyncSyslogQueueApi] is implemented.
401/// 
402/// ```ignore
403/// let log = 
404///     QueuedSyslog::openlog(
405///         Some("test1"), 
406///         LogStat::LOG_CONS | LogStat::LOG_NDELAY | LogStat::LOG_PID, 
407///         LogFacility::LOG_DAEMON,
408///         SyslogLocal::new()
409///     );
410/// ```
411/// 
412/// ```ignore
413/// let log = 
414///     QueuedSyslog
415///         ::<DefaultQueueAdapter, DefaultSyslogFormatter, SyslogLocal>
416///         ::openlog_with(
417///             Some("test1"), 
418///             LogStat::LOG_CONS | LogStat::LOG_NDELAY | LogStat::LOG_PID, 
419///             LogFacility::LOG_DAEMON,
420///             SyslogLocal::new()
421///         );
422/// ```
423/// 
424/// ```ignore
425/// pub static SYSLOG3: LazyLock<SyncSyslog<DefaultSyslogFormatter, SyslogLocal,>> = 
426///     LazyLock::new(|| 
427///         {
428///             QueuedSyslog
429///                 ::<DefaultQueueAdapter, DefaultSyslogFormatter, SyslogLocal>
430///                 ::openlog_with(
431///                     Some("test1"), 
432///                     LogStat::LOG_CONS | LogStat::LOG_NDELAY | LogStat::LOG_PID, 
433///                     LogFacility::LOG_DAEMON,
434///                     SyslogLocal::new()
435///                 )
436///                 .unwrap()
437///         }
438///     );
439/// ```
440/// 
441/// A stream is availble via [SyStreamApi].
442/// 
443/// ```ignore
444/// let _ = write!(SYSLOG.stream(Priority::LOG_DEBUG), "test {} 123 stream test ", d);
445/// ```
446/// 
447/// # Generics
448/// 
449/// * `S` - a [SyslogQueueChannel] a MPSC provider.
450/// 
451/// * `F` - a [SyslogFormatter] which sets the instance which would 
452///     format the message.
453/// 
454/// * `D` - a [SyslogDestination] instance which is either:
455///     [SyslogLocal], [SyslogFile], [SyslogNet], [SyslogTls]. By
456///     default a `SyslogLocal` is selected.
457#[derive(Debug, Clone)]
458pub struct QueuedSyslog<S = DefaultQueueAdapter, F = DefaultSyslogFormatter, D = DefaultLocalSyslogDestination>
459where F: SyslogFormatter, D: SyslogDestination, S: SyslogQueueChannel<F, D>
460{   
461    /// Control flag
462    run_control: Weak<AtomicBool>,
463
464    /// commands channel
465    pub(crate) tasks: S::ChannelSnd,//q_adapter::SendChannelSyCmd<D>,
466
467    /// process thread
468    thread: Arc<Mutex<Option<thread::JoinHandle<()>>>>,
469
470    /// phantom for [SyslogFormatter]
471    _p: PhantomData<F>,
472
473    /// phantom for [SyslogDestination]
474    _p2: PhantomData<D>,
475}
476
477
478unsafe impl<F, D, S> Send for QueuedSyslog<S, F, D> 
479where F: SyslogFormatter, D: SyslogDestination, S: SyslogQueueChannel<F, D>
480{}
481
482
483impl<F, D, S> Drop for QueuedSyslog<S, F, D>
484where F: SyslogFormatter, D: SyslogDestination, S: SyslogQueueChannel<F, D>
485{
486    fn drop(&mut self) 
487    {
488        if let Some(ctrl) = self.run_control.upgrade()
489        {   
490            ctrl.store(false, Ordering::SeqCst);
491
492            if let Err(_e) = self.tasks.q_send_blocking(SyCmd::form_stop())
493            {
494
495            }
496
497            let join_handle = self.thread.lock().unwrap().take().unwrap();
498
499            let _ = join_handle.join();
500        }
501    }
502}
503
504impl QueuedSyslog
505{
506    /// Opens a default connection to the local syslog server with default formatter with
507    /// formatter [SyslogFormatter] and destination [SyslogLocal].
508    /// 
509    /// # Arguments
510    /// 
511    /// * `ident` - A program name which will appear on the logs. If none, will be determined
512    ///     automatically.
513    /// 
514    /// * `logstat` - [LogStat] an instance config.
515    /// 
516    /// * `facility` - [LogFacility] a syslog facility.
517    /// 
518    /// * `net_tap_prov` - a [SyslogLocal] instance with configuration.
519    /// 
520    /// # Returns
521    /// 
522    /// A [SyRes] is returned ([Result]) with: 
523    /// 
524    /// * [Result::Ok] - with instance
525    /// 
526    /// * [Result::Err] - with error description.
527    pub 
528    fn openlog(ident: Option<&str>, logstat: LogStat, facility: LogFacility, net_tap_prov: DefaultLocalSyslogDestination) -> SyRes<Self>
529    {
530        // creating internal syslog struct
531
532         let log_items = 
533                LogItems::new(ident, 0xff, logstat, facility);
534
535        let stream = 
536            SyslogSocketLockless::<DefaultLocalSyslogDestination>::new(logstat, net_tap_prov)?;
537
538        let (inst, sender, run_ctrl) = 
539            SyslogInternal
540                ::<DefaultSyslogFormatter, DefaultLocalSyslogDestination, DefaultQueueAdapter>
541                ::new(log_items, stream)?;
542        
543        
544        let thr_name: String = "syslog_queue/0".into();
545
546        // initiate a thread
547        let thread_hnd = 
548            thread::Builder::new()
549                .name(thr_name.clone())
550                .spawn(move || 
551                    SyslogInternal
552                        ::<DefaultSyslogFormatter, DefaultLocalSyslogDestination, DefaultQueueAdapter>
553                        ::thread_worker(inst)
554                )
555                .map_err(|e| 
556                    map_error!("{} thread spawn failed. {}", thr_name, e)
557                )?;
558
559        // creating a syslog public struct instance
560        let ret = 
561            Self
562            {
563                run_control: run_ctrl,
564                tasks: sender,
565                thread: Arc::new(Mutex::new(Some(thread_hnd))),
566                _p: PhantomData,
567                _p2: PhantomData
568            };
569
570        return Ok(ret);
571    }
572}
573
574impl<F, D, S> QueuedSyslog<S, F, D>
575where F: SyslogFormatter, D: SyslogDestination, S: SyslogQueueChannel<F, D>
576{
577    /// Opens a default connection to the local syslog server with default formatter with
578    /// provided generics.
579    /// 
580    /// # Arguments
581    /// 
582    /// * `ident` - A program name which will appear on the logs. If none, will be determined
583    ///     automatically.
584    /// 
585    /// * `logstat` - [LogStat] an instance config.
586    /// 
587    /// * `facility` - [LogFacility] a syslog facility.
588    /// 
589    /// * `net_tap_prov` - a [SyslogLocal] instance with configuration.
590    /// 
591    /// # Returns
592    /// 
593    /// A [SyRes] is returned ([Result]) with: 
594    /// 
595    /// * [Result::Ok] - with instance
596    /// 
597    /// * [Result::Err] - with error description.
598    pub 
599    fn openlog_with(ident: Option<&str>, logstat: LogStat, facility: LogFacility, net_tap_prov: D) -> SyRes<QueuedSyslog<S, F, D>>
600    {
601        // creating internal syslog struct
602
603         let log_items = 
604                LogItems::new(ident, 0xff, logstat, facility);
605
606        let stream = 
607            SyslogSocketLockless::<D>::new(logstat, net_tap_prov)?;
608
609        let (inst, sender, run_ctrl) = 
610            SyslogInternal::<F, D, S>::new(log_items, stream)?;
611        
612        
613        let thr_name: String = "syslog_queue/0".into();
614
615        // initiate a thread
616        let thread_hnd = 
617            thread::Builder::new()
618                .name(thr_name.clone())
619                .spawn(move || SyslogInternal::<F, D, S>::thread_worker(inst))
620                .map_err(|e| 
621                    map_error!("{} thread spawn failed. {}", thr_name, e)
622                )?;
623
624        // creating a syslog public struct instance
625        let ret = 
626            Self
627            {
628                run_control: run_ctrl,
629                tasks: sender,
630                thread: Arc::new(Mutex::new(Some(thread_hnd))),
631                _p: PhantomData::<F>,
632                _p2: PhantomData::<D>,
633            };
634
635        return Ok(ret);
636    }
637}
638
639
640impl<F: SyslogFormatter, D: SyslogDestination, S: SyslogQueueChannel<F, D>> QueuedSyslog<S, F, D>
641{
642    /// Connects the current instance to the syslog server (destination).
643    pub 
644    fn connectlog(&self) -> SyRes<()>
645    {
646        let (sy_cmd, loopback) = 
647            SyCmd::form_connectlog();
648
649        self.tasks.q_send_blocking(sy_cmd)?;
650
651        return 
652            loopback
653                .recv_once_blocking()?;
654    }
655
656    /// Sets the logmask to filter out the syslog calls.
657    /// 
658    /// See macroses [LOG_MASK] and [LOG_UPTO] to generate mask
659    ///
660    /// # Example
661    ///
662    /// LOG_MASK!(Priority::LOG_EMERG) | LOG_MASK!(Priority::LOG_ERROR)
663    ///
664    /// or
665    ///
666    /// ~(LOG_MASK!(Priority::LOG_INFO))
667    /// LOG_UPTO!(Priority::LOG_ERROR)
668    pub 
669    fn setlogmask(&self, logmask: i32) -> SyRes<i32> 
670    {
671        let (sy_cmd, loopback) = 
672            SyCmd::form_logmask(logmask);
673
674        self.tasks.q_send_blocking(sy_cmd)?;
675
676        return 
677            loopback
678                .recv_once_blocking();
679    }
680
681    /// Closes connection to the syslog server (destination).
682    pub 
683    fn closelog(&self) -> SyRes<()> 
684    {
685        let (sy_cmd, loopback) = 
686            SyCmd::form_disconnectlog();
687
688        // send stop
689        self.tasks.q_send_blocking(sy_cmd)?;
690
691        return 
692            loopback
693                .recv_once_blocking()?;
694    }
695
696    /// Similar to libc, syslog() sends data to syslog server.
697    /// 
698    /// # Arguments
699    ///
700    /// * `pri` - a priority [Priority]
701    ///
702    /// * `fmt` - a formatter [SyslogFormatter] message. In C exists a functions with
703    ///     variable argumets amount. In Rust you should create your
704    ///     own macros like format!() or use format!()]. The [String] and ref `'static` 
705    ///     [str] can be passed directly.
706    /// 
707    /// # Returns 
708    /// 
709    /// A [SyRes] is returned which may describe an error.
710    pub 
711    fn syslog(&self, pri: Priority, fmt: F) -> SyRes<()>
712    {
713        // even if the thread is in a process of termination, there is
714        // no need to sync access to the run_control field as even if
715        // syslog thread will terminate before someone push something on the
716        // queue, it will be left in the queue until the end of program's time.
717        
718        let sy_cmd = SyCmd::form_syslog(pri, fmt);
719
720        return self.tasks.q_send_blocking(sy_cmd);
721    }
722
723    /// This function can be used to update the facility name, for example
724    /// after fork().
725    /// 
726    /// # Arguments
727    /// 
728    /// * `ident` - an [Option] optional new identity (up to 48 UTF8 chars)
729    ///     If set to [Option::None] would request the program name from OS.
730    pub 
731    fn change_identity(&self, ident: Option<&str>) -> SyRes<()>
732    {
733        let sy_cmd = 
734            SyCmd::form_change_ident(ident.map(|v| v.to_string()));
735
736        return 
737            self.tasks.q_send_blocking(sy_cmd);
738    }
739
740    /// Re-opens the connection to the syslog server. Can be used to 
741    /// rotate logs(handle SIGHUP).
742    /// 
743    /// # Returns
744    /// 
745    /// A [Result] is retured as [SyRes].
746    /// 
747    /// * [Result::Ok] - with empty inner type.
748    /// 
749    /// * [Result::Err] - an error code and description 
750    pub 
751    fn reconnect(&self) -> SyRes<()>
752    {
753        return 
754            self.tasks.q_send_blocking(SyCmd::form_reconnect());
755    }
756
757    /// Updates the instance's socket. `tap_data` [TapTypeData] should be of
758    /// the same variant (type) as current.
759    pub 
760    fn update_tap_data(&self, tap_data: D) -> SyRes<()>
761    {
762        let (tap_data_cmd, loopback) = SyCmd::form_update_tap(tap_data);
763
764        self.tasks.q_send_blocking(tap_data_cmd)?;
765
766        return 
767            loopback
768                .recv_once_blocking()?;
769    }
770}
771
772impl<F, D, S> QueuedSyslog<S, F, D>
773where F: SyslogFormatter, D: SyslogDestination, S: SyslogQueueChannel<F, D>
774{
775    /// Returns the streamable [SyStream] instance which can be used with [write!].
776    /// 
777    /// It implements both [std::fmt::Write] and [std::io::Write].
778    /// 
779    /// # Example
780    /// 
781    /// ```ignore
782    /// let log = 
783    ///     SingleSyslog::openlog(
784    ///         Some("test1"), 
785    ///         LogStat::LOG_CONS | LogStat::LOG_NDELAY | LogStat::LOG_PID, 
786    ///         LogFacility::LOG_DAEMON,
787    ///         SyslogLocal::new()
788    ///     ).unwrap();
789    /// 
790    /// write!(log.get_stream::<SyStreamPriDebug>(), "test stream singlesyslog {}", i).unwrap();
791    /// ```
792    pub 
793    fn get_stream<'t, PRI>(&'t self) -> SyStream<'t, PRI, D, F, &'t Self>
794    where PRI: SyStreamPri
795    {
796        SyStream
797        {
798            s: Some(self),
799            _p: PhantomData,
800            _p1: PhantomData,
801            _p2: PhantomData
802        }
803    }
804}
805
806
807impl<F, D, S> SyStreamSyslogApi<F, D>  
808for &QueuedSyslog<S, F, D>
809where F: SyslogFormatter, D: SyslogDestination, S: SyslogQueueChannel<F, D>
810{
811    type SYSLOG<'t> = &'t QueuedSyslog<S, F, D>;
812
813    fn syslog<'t>(syslog: Self::SYSLOG<'t>, pri: Priority, fmt: F) -> SyRes<()>
814    {
815        syslog.syslog(pri, fmt)
816    }
817}
818
819
820/// A queued implementation for ASYNC.
821#[cfg(all(feature = "build_with_queue", feature = "async_enabled"))]
822pub mod syslog_async_queue
823{
824    use crate::error::SyRes;
825    use crate::sy_sync_queue::{SyCmd, SyslogQueueChanSnd, SyslogQueueChannel, SyslogQueueOneChanRcv};
826    use crate::Priority;
827    use crate::{formatters::SyslogFormatter, SyslogDestination, QueuedSyslog};
828    use crate::a_sync::syslog_trait::AsyncSyslogQueueApi;
829
830    impl<F: SyslogFormatter, D: SyslogDestination, S: SyslogQueueChannel<F, D>> AsyncSyslogQueueApi<F, D> 
831    for QueuedSyslog<S, F, D>
832    {
833        async 
834        fn a_connectlog(&mut self) -> SyRes<()> 
835        {
836            let (sy_cmd, loopback) = 
837                SyCmd::form_connectlog();
838
839            self.tasks.q_send(sy_cmd).await?;
840
841            return 
842                loopback
843                    .recv_once()
844                    .await?;
845        }
846
847        /// Sets the logmask to filter out the syslog calls. This function behaves 
848        /// differently as it behaves in syslog_sync.rs or syslog_async.rs.
849        /// It may return an error if: syslog thread had exit and some thread calls
850        /// this function. Or something happened with channel. 
851        /// This function blocks until the previous mask is received.
852        /// 
853        /// See macroses [LOG_MASK] and [LOG_UPTO] to generate mask
854        ///
855        /// # Example
856        ///
857        /// LOG_MASK!(Priority::LOG_EMERG) | LOG_MASK!(Priority::LOG_ERROR)
858        ///
859        /// or
860        ///
861        /// ~(LOG_MASK!(Priority::LOG_INFO))
862        /// LOG_UPTO!(Priority::LOG_ERROR) 
863        async 
864        fn a_setlogmask(&self, logmask: i32) -> SyRes<i32>
865        {
866            let (sy_cmd, loopback) = 
867                SyCmd::form_logmask(logmask);
868
869            self.tasks.q_send(sy_cmd).await?;
870
871            return 
872                loopback
873                    .recv_once()
874                    .await;
875        }
876
877        /// Closes connection to the syslog server
878        async 
879        fn a_closelog(&self) -> SyRes<()>
880        {
881            let (sy_cmd, loopback) = 
882                SyCmd::form_disconnectlog();
883
884            // send stop
885            self.tasks.q_send(sy_cmd).await?;
886
887            return 
888                loopback
889                    .recv_once()
890                    .await?;
891        }
892
893        /// Similar to libc, syslog() sends data to syslog server, but asynchroniously.
894        /// 
895        /// # Arguments
896        ///
897        /// * `pri` - a priority [Priority]
898        ///
899        /// * `fmt` - a program's message to be sent as payload. The message is encoded with the
900        ///     [SyslogFormatter] and may be different for different formatters.
901        #[inline]
902        async 
903        fn a_syslog(&self, pri: Priority, fmt: F) -> SyRes<()>
904        {
905            let sy_cmd = SyCmd::form_syslog(pri, fmt);
906
907            return self.tasks.q_send(sy_cmd).await;
908        }
909
910
911        /// Performs the reconnection to the syslog server or file re-open.
912        /// 
913        /// # Returns
914        /// 
915        /// A [Result] is retured as [SyRes].
916        /// 
917        /// * [Result::Ok] - with empty inner type.
918        /// 
919        /// * [Result::Err] - an error code and description
920        async 
921        fn a_reconnect(&self) -> SyRes<()>
922        {
923            return 
924                self.tasks.q_send(SyCmd::form_reconnect()).await;
925        }
926
927        async 
928        fn a_change_identity(&self, ident: &str) -> SyRes<()> 
929        {
930            let sy_cmd = 
931                SyCmd::form_change_ident(Some(ident.to_string()));
932
933            return 
934                self.tasks.q_send(sy_cmd).await;
935        }
936
937        /// Updates the inner instance destionation i.e path to file
938        /// or server address. The type of destination can not be changed.
939        /// 
940        /// This function disconnects from syslog server if previously was 
941        /// connected (and reconnects if was connected previously).
942        /// 
943        /// # Arguments 
944        /// 
945        /// * `new_tap` - a consumed instance of type `D` [SyslogDestination]
946        /// 
947        /// # Returns 
948        /// 
949        /// A [SyRes] is returned. An error may be returned if:
950        /// 
951        /// * connection to server was failed
952        /// 
953        /// * incorrect type
954        /// 
955        /// * disconnect frm server failed
956        async 
957        fn a_update_tap_data(&self, new_tap: D) -> SyRes<()>
958        {
959            let (tap_data_cmd, loopback) = SyCmd::form_update_tap(new_tap);
960
961            self.tasks.q_send(tap_data_cmd).await?;
962
963            return 
964                loopback
965                    .recv_once()
966                    .await?;
967        }
968    }
969}
970*/
971