acta 0.1.0

Make Tracing Great Again.
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
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126

#[cfg(feature = "custom-async")]
use std::io::Write;
use std::sync::Arc;
#[cfg(feature = "custom-async")]
use std::sync::atomic::{AtomicUsize, Ordering};

#[cfg(feature = "custom-async")]
use tokio::io::AsyncWrite;
#[cfg(feature = "custom-async")]
use tokio::io::{AsyncWriteExt, stderr, stdout};
#[cfg(feature = "custom-async")]
use tokio::sync::mpsc;
use tracing_subscriber::fmt::MakeWriter;

#[derive(Clone, Copy, Debug)]
#[allow(clippy::exhaustive_enums)]
pub enum AsyncWriterTarget {
    Stdout,
    Stderr,
}

#[cfg(feature = "custom-async")]
#[derive(Clone, Debug)]
#[allow(clippy::module_name_repetitions)]
pub struct AsyncWriter {
    sender: mpsc::UnboundedSender<Vec<u8>>,
    count: Arc<AtomicUsize>,
}

#[cfg(feature = "custom-async")]
impl AsyncWriter {
    pub fn pending_writes(&self) -> usize {
        self.count.load(Ordering::Relaxed)
    }
}

#[cfg(feature = "custom-async")]
impl Write for AsyncWriter {
    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
        self.count.fetch_add(1, Ordering::Relaxed);
        self.sender.send(buf.to_vec()).map_err(|_| {
            self.count.fetch_sub(1, Ordering::Relaxed);
            std::io::Error::new(std::io::ErrorKind::BrokenPipe, "async writer closed")
        })?;
        Ok(buf.len())
    }

    fn flush(&mut self) -> std::io::Result<()> {
        Ok(())
    }
}

#[cfg(feature = "custom-async")]
impl MakeWriter<'_> for AsyncWriter {
    type Writer = Self;

    fn make_writer(&self) -> Self::Writer {
        self.clone()
    }
}

#[cfg(feature = "custom-async")]
#[allow(clippy::module_name_repetitions)]
pub fn async_writer() -> AsyncWriter {
    async_writer_for(AsyncWriterTarget::Stdout)
}

#[cfg(feature = "custom-async")]
pub fn async_writer_for(target: AsyncWriterTarget) -> AsyncWriter {
    let (sender, mut receiver) = mpsc::unbounded_channel::<Vec<u8>>();
    let count = Arc::new(AtomicUsize::new(0));
    let count_clone = count.clone();

    tokio::spawn(async move {
        let writer: &mut (dyn AsyncWrite + Unpin + Send) = match target {
            AsyncWriterTarget::Stdout => &mut stdout(),
            AsyncWriterTarget::Stderr => &mut stderr(),
        };

        while let Some(data) = receiver.recv().await {
            if let Err(e) = writer.write_all(&data).await {
                let _unused = writeln!(std::io::stderr(), "async writer error: {e}");
            }
            count_clone.fetch_sub(1, Ordering::Relaxed);
        }
    });

    AsyncWriter { sender, count }
}

#[cfg(feature = "native-async")]
#[allow(clippy::module_name_repetitions)]
pub struct NativeAsyncWriter {
    writer: tracing_appender::non_blocking::NonBlocking,
    _guard: tracing_appender::non_blocking::WorkerGuard,
}

#[cfg(feature = "native-async")]
impl std::fmt::Debug for NativeAsyncWriter {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("NativeAsyncWriter").finish_non_exhaustive()
    }
}

#[cfg(feature = "native-async")]
impl MakeWriter<'_> for NativeAsyncWriter {
    type Writer = tracing_appender::non_blocking::NonBlocking;

    fn make_writer(&self) -> Self::Writer {
        self.writer.to_owned()
    }
}

#[cfg(feature = "native-async")]
#[allow(clippy::module_name_repetitions)]
pub fn native_async_writer(target: AsyncWriterTarget) -> NativeAsyncWriter {
    let (writer, guard) = match target {
        AsyncWriterTarget::Stdout => tracing_appender::non_blocking(std::io::stdout()),
        AsyncWriterTarget::Stderr => tracing_appender::non_blocking(std::io::stderr()),
    };

    NativeAsyncWriter {
        writer,
        _guard: guard,
    }
}