logforth-append-async 0.3.0

Asynchronous appender for Logforth.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84

// Copyright 2024 FastLabs Developers
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

use std::sync::Arc;
use std::thread::JoinHandle;

use arc_swap::ArcSwapOption;
use crossbeam_channel::Sender;
use logforth_core::Error;

use crate::Overflow;
use crate::Task;

#[derive(Debug)]
pub(crate) struct AsyncState(ArcSwapOption<State>);

#[derive(Debug)]
struct State {
    sender: Sender<Task>,
    handle: JoinHandle<()>,
}

impl AsyncState {
    pub(crate) fn new(sender: Sender<Task>, handle: JoinHandle<()>) -> Self {
        let state = State { sender, handle };
        Self(ArcSwapOption::from(Some(Arc::new(state))))
    }

    pub(crate) fn send_task(&self, task: Task, overflow: Overflow) -> Result<(), Error> {
        let state = self.0.load();
        // SAFETY: state is always Some before dropped.
        let state = state.as_ref().unwrap();
        let sender = &state.sender;

        match overflow {
            Overflow::Block => sender.send(task).map_err(|err| {
                Error::new(match err.0 {
                    Task::Log { .. } => "failed to send log task to async appender",
                    Task::Flush { .. } => "failed to send flush task to async appender",
                })
            }),
            Overflow::DropIncoming => match sender.try_send(task) {
                Ok(()) => Ok(()),
                Err(crossbeam_channel::TrySendError::Full(_)) => Ok(()),
                Err(crossbeam_channel::TrySendError::Disconnected(task)) => {
                    Err(Error::new(match task {
                        Task::Log { .. } => "failed to send log task to async appender",
                        Task::Flush { .. } => "failed to send flush task to async appender",
                    }))
                }
            },
        }
    }

    pub(crate) fn destroy(&self) {
        if let Some(state) = self.0.swap(None) {
            // SAFETY: state has always one strong count before swapped.
            let State { sender, handle } = Arc::into_inner(state).unwrap();

            // drop our sender, threads will break the loop after receiving and processing
            drop(sender);

            // wait for the thread to finish
            handle.join().expect("failed to join async appender thread");
        }
    }
}

impl Drop for AsyncState {
    fn drop(&mut self) {
        self.destroy();
    }
}