rws_server/log/
tail.rs

1//! The last access-log lines in memory, and live followers of new ones,
2//! for the admin API (18.10.0). Off until [`enable`]: then each batch the
3//! access-log sink writes out (every 100 ms, off the request path) is also
4//! split into lines here.
5
6use std::collections::VecDeque;
7use std::sync::atomic::{AtomicBool, Ordering};
8use std::sync::{Mutex, OnceLock};
9
10use crate::sse::{SseEvent, SseSender};
11
12/// Lines kept for [`recent`].
13pub const CAPACITY: usize = 500;
14/// Followers at once; more are refused.
15pub const MAX_FOLLOWERS: usize = 8;
16/// A longer line is cut to this many bytes (on a character boundary).
17pub const MAX_LINE: usize = 8 * 1024;
18
19static ENABLED: AtomicBool = AtomicBool::new(false);
20
21#[derive(Default)]
22struct Tail {
23    lines: VecDeque<String>,
24    followers: Vec<SseSender>,
25}
26
27fn tail() -> &'static Mutex<Tail> {
28    static TAIL: OnceLock<Mutex<Tail>> = OnceLock::new();
29    TAIL.get_or_init(Mutex::default)
30}
31
32/// Start keeping lines (the admin listener does, at start).
33pub fn enable() {
34    ENABLED.store(true, Ordering::SeqCst);
35}
36
37#[doc(hidden)]
38pub fn enabled() -> bool {
39    ENABLED.load(Ordering::Relaxed)
40}
41
42fn cut(line: &str) -> String {
43    if line.len() <= MAX_LINE {
44        return line.to_string();
45    }
46    let mut end = MAX_LINE;
47    while !line.is_char_boundary(end) {
48        end -= 1;
49    }
50    line[..end].to_string()
51}
52
53/// Lines the sink wrote out, newline-separated.
54#[doc(hidden)]
55pub fn record(bytes: &[u8]) {
56    if !enabled() {
57        return;
58    }
59    let text = String::from_utf8_lossy(bytes);
60    let mut tail = tail().lock().unwrap_or_else(|e| e.into_inner());
61    for line in text.split('\n').filter(|l| !l.is_empty()) {
62        let line = cut(line);
63        // A follower that's gone is dropped; one that's behind misses lines.
64        tail.followers.retain(|f| f.try_send(SseEvent::data(line.clone())).is_ok());
65        if tail.lines.len() == CAPACITY {
66            tail.lines.pop_front();
67        }
68        tail.lines.push_back(line);
69    }
70}
71
72/// The last `n` lines, oldest first.
73pub fn recent(n: usize) -> Vec<String> {
74    let tail = tail().lock().unwrap_or_else(|e| e.into_inner());
75    tail.lines.iter().skip(tail.lines.len().saturating_sub(n)).cloned().collect()
76}
77
78/// Add a follower: `false` when [`MAX_FOLLOWERS`] already follow.
79pub fn follow(sender: SseSender) -> bool {
80    let mut tail = tail().lock().unwrap_or_else(|e| e.into_inner());
81    tail.followers.retain(|f| f.try_send(SseEvent::data(String::new()).event_type("ping")).is_ok());
82    if tail.followers.len() >= MAX_FOLLOWERS {
83        return false;
84    }
85    tail.followers.push(sender);
86    true
87}
rws_server/log/tail.rs

rws_server/log/
tail.rs
