use std::fs::{File, OpenOptions};
use std::io::{BufWriter, Write};
use std::path::Path;
use std::sync::Mutex;
use super::{CachePadded, Sink, lock_recover, 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,
}
}
}
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,
}
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
}
pub fn open(self) -> Result<FileSink<F>> {
let file = OpenOptions::new()
.create(true)
.write(true)
.append(self.append)
.truncate(!self.append)
.open(&self.path)?;
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(_)));
}
}