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
use std::{collections::BTreeMap, io, io::Write, pin::Pin, sync::Arc};
use tokio::{io::AsyncWrite, sync::Mutex};
use crate::{log_failure, Key, Value, Writer};
pub struct AsyncJSONWriter<W: AsyncWrite + Sync + Send + 'static>(Arc<Mutex<Pin<Box<W>>>>);
impl<W: AsyncWrite + Sync + Send + 'static> AsyncJSONWriter<W> {
pub fn new(w: W) -> Self {
Self(Arc::new(Mutex::new(Box::pin(w))))
}
}
impl<W: AsyncWrite + Sync + Send + 'static> Writer for AsyncJSONWriter<W> {
fn write_log(&self, value: &BTreeMap<Key, Value>) -> Result<(), io::Error> {
let mut buf = Vec::with_capacity(256);
serde_json::to_writer(&mut buf, value).map_err(io::Error::from)?;
buf.write_all(b"\n").map_err(io::Error::from)?;
let w = self.0.clone();
tokio::spawn(async move {
use tokio::io::AsyncWriteExt;
let mut w = w.lock().await;
if let Err(err) = w.as_mut().write_all(&buf).await {
log_failure(format!("AsyncJSONWriter failed to write log: {}", err).as_str());
}
});
Ok(())
}
}
pub fn new_writer<W: AsyncWrite + Sync + Send + 'static>(w: W) -> Box<dyn Writer> {
Box::new(AsyncJSONWriter::new(w))
}