use std::fs::File;
use std::io::{BufWriter, Write};
use std::path::Path;
use std::sync::Mutex;
use super::{CachePadded, Sink, lock_recover, open_log_file, write_locked};
use crate::error::{Error, Result};
use crate::format::Format;
use crate::record::Record;
const DEFAULT_BUFFER: usize = 8 * 1024;
#[cfg_attr(all(feature = "json"), doc = "```no_run")]
#[cfg_attr(not(all(feature = "json")), doc = "```ignore")]
pub struct FileSink<F: Format> {
format: F,
target: CachePadded<Mutex<BufWriter<File>>>,
sync_on_flush: bool,
}
impl<F: Format> FileSink<F> {
pub fn append<P: AsRef<Path>>(path: P, format: F) -> Result<Self> {
Self::builder(path, format).append(true).open()
}
pub fn create<P: AsRef<Path>>(path: P, format: F) -> Result<Self> {
Self::builder(path, format).append(false).open()
}
pub fn builder<P: AsRef<Path>>(path: P, format: F) -> FileSinkBuilder<F> {
FileSinkBuilder {
path: path.as_ref().to_path_buf(),
format,
append: true,
buffer_capacity: DEFAULT_BUFFER,
sync_on_flush: false,
mode: None,
follow_symlinks: false,
}
}
}
impl<F: Format> Sink for FileSink<F> {
fn write_record(&self, record: &Record<'_>) -> Result<()> {
write_locked(&self.target, &self.format, record)
}
fn flush(&self) -> Result<()> {
let mut file = lock_recover(&self.target);
file.flush()?;
if self.sync_on_flush {
file.get_ref().sync_data().map_err(Error::from)?;
}
Ok(())
}
}
impl<F: Format + core::fmt::Debug> core::fmt::Debug for FileSink<F> {
fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
f.debug_struct("FileSink")
.field("format", &self.format)
.field("sync_on_flush", &self.sync_on_flush)
.finish_non_exhaustive()
}
}
#[derive(Debug)]
#[must_use = "call `open()` to create the sink"]
pub struct FileSinkBuilder<F> {
path: std::path::PathBuf,
format: F,
append: bool,
buffer_capacity: usize,
sync_on_flush: bool,
mode: Option<u32>,
follow_symlinks: bool,
}
impl<F: Format> FileSinkBuilder<F> {
pub const fn append(mut self, append: bool) -> Self {
self.append = append;
self
}
pub const fn buffer_capacity(mut self, bytes: usize) -> Self {
self.buffer_capacity = bytes;
self
}
pub const fn sync_on_flush(mut self, enabled: bool) -> Self {
self.sync_on_flush = enabled;
self
}
#[cfg_attr(all(feature = "json"), doc = "```no_run")]
#[cfg_attr(not(all(feature = "json")), doc = "```ignore")]
pub const fn mode(mut self, mode: u32) -> Self {
self.mode = Some(mode);
self
}
pub const fn follow_symlinks(mut self, follow: bool) -> Self {
self.follow_symlinks = follow;
self
}
pub fn open(self) -> Result<FileSink<F>> {
let file = open_log_file(&self.path, self.append, self.mode, self.follow_symlinks)?;
Ok(FileSink {
format: self.format,
target: CachePadded(Mutex::new(BufWriter::with_capacity(
self.buffer_capacity,
file,
))),
sync_on_flush: self.sync_on_flush,
})
}
}
#[cfg(all(test, feature = "json"))]
mod tests {
use super::*;
use crate::format::JsonFormat;
use crate::{Level, Metadata};
fn temp_path(name: &str) -> std::path::PathBuf {
let dir = std::env::temp_dir().join(format!("log-io-file-{}-{name}", std::process::id()));
let _ = std::fs::remove_file(&dir);
dir
}
fn record(msg: &str) -> Record<'_> {
Record::new(Metadata::new(Level::Info, "t"), msg, &[])
}
#[test]
fn test_file_sink_append_preserves_existing_content() {
let path = temp_path("append");
std::fs::write(&path, "existing\n").unwrap();
let sink = FileSink::append(&path, JsonFormat::new()).unwrap();
sink.write_record(&record("one")).unwrap();
sink.flush().unwrap();
let text = std::fs::read_to_string(&path).unwrap();
assert!(text.starts_with("existing\n"), "{text}");
assert!(text.contains("\"message\":\"one\""), "{text}");
let _ = std::fs::remove_file(&path);
}
#[test]
fn test_file_sink_create_truncates() {
let path = temp_path("create");
std::fs::write(&path, "old content\n").unwrap();
let sink = FileSink::create(&path, JsonFormat::new()).unwrap();
sink.write_record(&record("fresh")).unwrap();
sink.shutdown().unwrap();
let text = std::fs::read_to_string(&path).unwrap();
assert!(!text.contains("old content"), "{text}");
assert!(text.contains("fresh"), "{text}");
let _ = std::fs::remove_file(&path);
}
#[test]
fn test_file_sink_buffers_until_flush() {
let path = temp_path("buffered");
let sink = FileSink::create(&path, JsonFormat::new()).unwrap();
sink.write_record(&record("pending")).unwrap();
assert_eq!(std::fs::read(&path).unwrap().len(), 0, "still buffered");
sink.flush().unwrap();
assert!(!std::fs::read(&path).unwrap().is_empty());
let _ = std::fs::remove_file(&path);
}
#[test]
fn test_file_sink_unbuffered_writes_through() {
let path = temp_path("unbuffered");
let sink = FileSink::builder(&path, JsonFormat::new())
.append(false)
.buffer_capacity(0)
.sync_on_flush(true)
.open()
.unwrap();
sink.write_record(&record("now")).unwrap();
assert!(
std::fs::read_to_string(&path).unwrap().contains("now"),
"unbuffered sink must write before returning"
);
sink.flush().unwrap();
assert!(format!("{sink:?}").contains("sync_on_flush: true"));
let _ = std::fs::remove_file(&path);
}
#[test]
fn test_file_sink_open_error_is_reported() {
let dir = std::env::temp_dir().join(format!("log-io-missing-{}", std::process::id()));
let path = dir.join("nested").join("file.log");
let err = FileSink::append(&path, JsonFormat::new()).unwrap_err();
assert!(matches!(err, Error::Io(_)));
}
#[cfg(unix)]
#[test]
fn test_file_sink_mode_applies_to_a_new_file() {
use std::os::unix::fs::PermissionsExt;
let path = temp_path("mode.jsonl");
let sink = FileSink::builder(&path, JsonFormat::new())
.mode(0o600)
.open()
.unwrap();
drop(sink);
let mode = std::fs::metadata(&path).unwrap().permissions().mode() & 0o777;
assert_eq!(mode, 0o600);
let _ = std::fs::remove_file(&path);
}
#[cfg(unix)]
#[test]
fn test_file_sink_default_mode_is_owner_only() {
use std::os::unix::fs::PermissionsExt;
let path = temp_path("default-mode.jsonl");
let sink = FileSink::append(&path, JsonFormat::new()).unwrap();
drop(sink);
let mode = std::fs::metadata(&path).unwrap().permissions().mode() & 0o777;
assert_eq!(mode, 0o600);
let _ = std::fs::remove_file(&path);
}
#[test]
fn test_file_sink_refuses_a_symlink_and_leaves_its_target_alone() {
let victim = temp_path("victim.txt");
let link = temp_path("link.jsonl");
std::fs::write(&victim, "important=1\n").unwrap();
let _ = std::fs::remove_file(&link);
#[cfg(unix)]
let made = std::os::unix::fs::symlink(&victim, &link);
#[cfg(windows)]
let made = std::os::windows::fs::symlink_file(&victim, &link);
if made.is_err() {
let _ = std::fs::remove_file(&victim);
return;
}
assert!(FileSink::create(&link, JsonFormat::new()).is_err());
assert!(FileSink::append(&link, JsonFormat::new()).is_err());
assert_eq!(std::fs::read_to_string(&victim).unwrap(), "important=1\n");
let followed = FileSink::builder(&link, JsonFormat::new())
.follow_symlinks(true)
.open();
assert!(
followed.is_ok(),
"follow_symlinks(true) must allow the link"
);
drop(followed);
let _ = std::fs::remove_file(&link);
let _ = std::fs::remove_file(&victim);
}
#[test]
fn test_file_sink_refuses_a_directory() {
let dir = temp_path("a-directory");
let _ = std::fs::remove_dir(&dir);
std::fs::create_dir(&dir).unwrap();
assert!(FileSink::append(&dir, JsonFormat::new()).is_err());
let _ = std::fs::remove_dir(&dir);
}
#[test]
fn test_file_sink_create_truncates_a_regular_file() {
let path = temp_path("truncate.jsonl");
std::fs::write(&path, "old content that is long\n").unwrap();
let sink = FileSink::create(&path, JsonFormat::new()).unwrap();
sink.write_record(&record("new")).unwrap();
sink.flush().unwrap();
let text = std::fs::read_to_string(&path).unwrap();
assert!(!text.contains("old content"), "{text}");
assert!(text.contains("\"message\":\"new\""), "{text}");
let _ = std::fs::remove_file(&path);
}
#[test]
fn test_file_sink_without_mode_still_opens() {
let path = temp_path("no-mode.jsonl");
assert!(FileSink::builder(&path, JsonFormat::new()).open().is_ok());
let _ = std::fs::remove_file(&path);
}
}