rust_web_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
37pub(crate) fn enabled() -> bool {
38    ENABLED.load(Ordering::Relaxed)
39}
40
41fn cut(line: &str) -> String {
42    if line.len() <= MAX_LINE {
43        return line.to_string();
44    }
45    let mut end = MAX_LINE;
46    while !line.is_char_boundary(end) {
47        end -= 1;
48    }
49    line[..end].to_string()
50}
51
52/// Lines the sink wrote out, newline-separated.
53pub(crate) fn record(bytes: &[u8]) {
54    if !enabled() {
55        return;
56    }
57    let text = String::from_utf8_lossy(bytes);
58    let mut tail = tail().lock().unwrap_or_else(|e| e.into_inner());
59    for line in text.split('\n').filter(|l| !l.is_empty()) {
60        let line = cut(line);
61        // A follower that's gone is dropped; one that's behind misses lines.
62        tail.followers.retain(|f| f.try_send(SseEvent::data(line.clone())).is_ok());
63        if tail.lines.len() == CAPACITY {
64            tail.lines.pop_front();
65        }
66        tail.lines.push_back(line);
67    }
68}
69
70/// The last `n` lines, oldest first.
71pub fn recent(n: usize) -> Vec<String> {
72    let tail = tail().lock().unwrap_or_else(|e| e.into_inner());
73    tail.lines.iter().skip(tail.lines.len().saturating_sub(n)).cloned().collect()
74}
75
76/// Add a follower: `false` when [`MAX_FOLLOWERS`] already follow.
77pub fn follow(sender: SseSender) -> bool {
78    let mut tail = tail().lock().unwrap_or_else(|e| e.into_inner());
79    tail.followers.retain(|f| f.try_send(SseEvent::data(String::new()).event_type("ping")).is_ok());
80    if tail.followers.len() >= MAX_FOLLOWERS {
81        return false;
82    }
83    tail.followers.push(sender);
84    true
85}
rust_web_server/log/tail.rs

rust_web_server/log/
tail.rs
