From b759c7ec2ac5411f7dd254bb19a20c27a166e7e2 Mon Sep 17 00:00:00 2001 From: Max Verevkin Date: Mon, 4 Apr 2022 16:53:49 +0300 Subject: dont use line buffering when using json protocol --- src/i3bar_protocol.rs | 157 +++++++++++++++++++++++++------------------------- src/lines_buffer.rs | 54 ----------------- src/main.rs | 4 +- src/status_cmd.rs | 22 +++---- src/utils.rs | 105 +++++++++++++++++++++++++++++++++ 5 files changed, 194 insertions(+), 148 deletions(-) delete mode 100644 src/lines_buffer.rs create mode 100644 src/utils.rs diff --git a/src/i3bar_protocol.rs b/src/i3bar_protocol.rs index 6bc0533..4ef68ba 100644 --- a/src/i3bar_protocol.rs +++ b/src/i3bar_protocol.rs @@ -1,9 +1,9 @@ use crate::color::Color; use crate::pointer_btn::PointerBtn; use crate::text::Align; +use crate::utils::{de_first_json, de_last_json, last_line}; use serde::Deserialize; use serde::Serialize; -use serde_json::{Deserializer, Result as SerdeResult}; use std::io::{Error, ErrorKind, Result}; #[derive(Deserialize, Default)] @@ -80,63 +80,104 @@ pub struct JsonHeader { pub enum Protocol { Unknown, - PlainText(Option), + PlainText { + line: Option, + buf: Vec, + }, JsonNotStarted(JsonHeader), - Json(JsonHeader, Option>, Option), + Json { + header: JsonHeader, + blocks: Option>, + buf: Vec, + }, } impl Protocol { - pub fn process_line(&mut self, line: String) -> Result<()> { + pub fn process_new_bytes(&mut self, bytes: &[u8]) -> Result<()> { macro_rules! invalid { - ($fmt:expr) => { - return Err(Error::new(ErrorKind::InvalidData, format!($fmt))) + ($fmt:expr $(,$t:expr)*) => { + return Err(Error::new(ErrorKind::InvalidData, format!($fmt $(,$t)*))) }; } match self { Self::Unknown => { - if let Ok(header) = serde_json::from_str::(&line) { + if let Ok((Some(header), rem)) = de_first_json::(bytes) { if header.version == 1 { - *self = Protocol::JsonNotStarted(header); - return Ok(()); + *self = Self::JsonNotStarted(header); + return self.process_new_bytes(rem); + } + } + *self = Self::PlainText { + line: None, + buf: Vec::new(), + }; + return self.process_new_bytes(bytes); + } + Self::PlainText { line, buf } => { + if buf.is_empty() { + if let Some((new_line, rem)) = last_line(bytes) { + *line = Some(String::from_utf8_lossy(new_line).into()); + buf.extend_from_slice(rem); + } else { + buf.extend_from_slice(bytes); + } + } else { + buf.extend_from_slice(bytes); + if let Some((new_line, rem)) = last_line(buf) { + *line = Some(String::from_utf8_lossy(new_line).into()); + let used = buf.len() - rem.len(); + buf.drain(..used); } } - *self = Self::PlainText(Some(line)) } - Self::PlainText(s) => *s = Some(line), Self::JsonNotStarted(header) => { - let line = line.trim_start(); - if !line.is_empty() { - if !line.starts_with('[') { - invalid!("Expected '['"); + let mut bytes = bytes; + while bytes.first().map_or(false, |&x| x == b' ' || x == b'\n') { + bytes = &bytes[1..]; + } + match bytes.first() { + Some(b'[') => { + *self = Self::Json { + header: *header, + blocks: None, + buf: Vec::new(), + }; + return self.process_new_bytes(&bytes[1..]); } - *self = Self::Json(*header, None, None); - self.process_line(line[1..].to_owned())?; + Some(other) => invalid!("invalid json: expected '[', got '{}'", *other as char), + _ => (), } } - Self::Json(_, blocks, start_orig) => match start_orig { - Some(start) => { - start.push_str(&line); - match de_last::>(start) { - Err(e) => invalid!("Invalid JSON: {e}"), - Ok((rem, new_blocks)) => { - *start_orig = (!rem.is_empty()).then(|| rem.to_owned()); + Self::Json { + header: _, + blocks, + buf, + } => { + if buf.is_empty() { + match de_last_json(bytes) { + Err(e) => invalid!("invalid json: {e}"), + Ok((new_blocks, rem)) => { if let Some(new_blocks) = new_blocks { *blocks = Some(new_blocks); } + buf.extend_from_slice(rem); } } - } - None => match de_last::>(&line) { - Err(e) => invalid!("Invalid JSON: {e}"), - Ok((rem, new_blocks)) => { - *start_orig = (!rem.is_empty()).then(|| rem.to_owned()); - if let Some(new_blocks) = new_blocks { - *blocks = Some(new_blocks); + } else { + buf.extend_from_slice(bytes); + match de_last_json(buf) { + Err(e) => invalid!("invalid json: {e}"), + Ok((new_blocks, rem)) => { + if let Some(new_blocks) = new_blocks { + *blocks = Some(new_blocks); + } + let used = buf.len() - rem.len(); + buf.drain(..used); } } - }, - }, + } + } } Ok(()) } @@ -144,62 +185,20 @@ impl Protocol { pub fn get_blocks(&mut self) -> Option> { match self { Self::Unknown => None, - Self::PlainText(text) => Some(vec![Block { - full_text: text.take()?, + Self::PlainText { line, .. } => Some(vec![Block { + full_text: line.take()?, ..Default::default() }]), Self::JsonNotStarted(_) => None, - Self::Json(_, blocks, _) => blocks.take(), + Self::Json { blocks, .. } => blocks.take(), } } pub fn supports_clicks(&self) -> bool { match self { Self::JsonNotStarted(h) => h.click_events, - Self::Json(h, _, _) => h.click_events, + Self::Json { header, .. } => header.click_events, _ => false, } } } - -/// Deserialize the last complete object. Returns (`remaining`, `object`). See tests for examples. -fn de_last<'a, T: Deserialize<'a>>(mut s: &'a str) -> SerdeResult<(&'a str, Option)> { - let mut last = None; - let mut tmp; - loop { - (s, tmp) = de_once(s)?; - last = match tmp { - Some(obj) => Some(obj), - None => return Ok((s, last)), - }; - } -} - -/// Deserialize the first complete object. Returns (`remaining`, `object`). -fn de_once<'a, T: Deserialize<'a>>(s: &'a str) -> SerdeResult<(&'a str, Option)> { - let s = s.trim_start_matches(|x: char| x.is_ascii_whitespace() || x == ','); - let mut de = Deserializer::from_str(s).into_iter(); - match de.next() { - Some(Ok(obj)) => Ok((&s[de.byte_offset()..], Some(obj))), - Some(Err(e)) if e.is_eof() => Ok((&s[de.byte_offset()..], None)), - Some(Err(e)) => Err(e), - None => Ok((&s[de.byte_offset()..], None)), - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn streaming_json() { - let s = ",[2], [3], [4, 3],[32][3] "; - assert_eq!(de_last::>(s).unwrap(), ("", Some(vec![3]))); - - let s = ",[2], [3], [4, 3],[32][3] [2, 4"; - assert_eq!(de_last::>(s).unwrap(), ("[2, 4", Some(vec![3]))); - - let s = ",[2], [3], [4, 3],[32] invalid"; - assert!(de_last::>(s).is_err()); - } -} diff --git a/src/lines_buffer.rs b/src/lines_buffer.rs deleted file mode 100644 index 7813ea2..0000000 --- a/src/lines_buffer.rs +++ /dev/null @@ -1,54 +0,0 @@ -use std::borrow::Cow; -use std::io::{BufRead, BufReader, Read, Result}; - -pub struct LinesBuffer { - reader: BufReader, - next_line: Vec, -} - -impl LinesBuffer { - pub fn new(inner: R) -> Self { - Self { - reader: BufReader::new(inner), - next_line: Vec::new(), - } - } - - pub fn fill_buf(&mut self) -> Result<()> { - self.reader.fill_buf()?; - Ok(()) - } - - pub fn inner(&self) -> &R { - self.reader.get_ref() - } -} - -impl Iterator for &mut LinesBuffer { - type Item = String; - - fn next(&mut self) -> Option { - let pending = self.reader.buffer(); - let (result, used) = match memchr::memchr(b'\n', pending) { - None => { - self.next_line.extend_from_slice(pending); - (None, pending.len()) - } - Some(i) => { - self.next_line.extend_from_slice(&pending[..i]); - let line = from_utf8(&self.next_line); - self.next_line.clear(); - (Some(line), i + 1) - } - }; - self.reader.consume(used); - result - } -} - -fn from_utf8(bytes: &[u8]) -> String { - match String::from_utf8_lossy(bytes) { - Cow::Borrowed(s) => s.to_owned(), - Cow::Owned(s) => s, - } -} diff --git a/src/main.rs b/src/main.rs index 1fe335e..48542d9 100644 --- a/src/main.rs +++ b/src/main.rs @@ -28,13 +28,13 @@ mod button_manager; mod color; mod config; mod i3bar_protocol; -mod lines_buffer; mod ord_adaptor; mod pointer_btn; mod river_protocols; mod status_cmd; mod tags; mod text; +mod utils; use button_manager::ButtonManager; use config::Config; @@ -247,7 +247,7 @@ fn main() { readable: true, writable: false, }, - calloop::Mode::Edge, + calloop::Mode::Level, ), move |ready, cmd, _| { info!("status command update"); diff --git a/src/status_cmd.rs b/src/status_cmd.rs index 53c3873..e77e4b2 100644 --- a/src/status_cmd.rs +++ b/src/status_cmd.rs @@ -1,13 +1,12 @@ use crate::Surface; use std::cell::RefCell; use std::io::Result; -use std::io::Write; +use std::io::{BufRead, BufReader, Write}; use std::os::unix::prelude::{AsRawFd, RawFd}; -use std::process::{Child, ChildStdin, ChildStdout, Command, Stdio}; +use std::process::{ChildStdin, ChildStdout, Command, Stdio}; use std::rc::Rc; use crate::i3bar_protocol::{Block, Event, Protocol}; -use crate::lines_buffer::LinesBuffer; use crate::pointer_btn::PointerBtn; #[derive(Clone)] @@ -16,9 +15,7 @@ pub struct StatusCmd { } struct Inner { - #[allow(dead_code)] - child: Child, - output: LinesBuffer, + output: BufReader, input: ChildStdin, protocol: Protocol, blocks: Rc>>, @@ -27,16 +24,16 @@ struct Inner { impl AsRawFd for StatusCmd { fn as_raw_fd(&self) -> RawFd { - self.inner.borrow().output.inner().as_raw_fd() + self.inner.borrow().output.get_ref().as_raw_fd() } } impl Inner { pub fn notify_available(&mut self) -> Result<()> { - self.output.fill_buf()?; - for line in &mut self.output { - self.protocol.process_line(line)?; - } + let buf = self.output.fill_buf()?; + self.protocol.process_new_bytes(buf)?; + let consumed = buf.len(); + self.output.consume(consumed); if let Some(new_blocks) = self.protocol.get_blocks() { *self.blocks.borrow_mut() = new_blocks; for s in &mut *self.surfaces.borrow_mut() { @@ -58,11 +55,10 @@ impl StatusCmd { .stdin(Stdio::piped()) .stdout(Stdio::piped()) .spawn()?; - let output = LinesBuffer::new(child.stdout.take().unwrap()); + let output = BufReader::new(child.stdout.take().unwrap()); let input = child.stdin.take().unwrap(); Ok(Self { inner: Rc::new(RefCell::new(Inner { - child, output, input, protocol: Protocol::Unknown, diff --git a/src/utils.rs b/src/utils.rs new file mode 100644 index 0000000..2ed1868 --- /dev/null +++ b/src/utils.rs @@ -0,0 +1,105 @@ +//! Some usefull functions + +use serde::Deserialize; +use serde_json::{Deserializer, Error as JsonError}; + +/// Retuns (`last_line`, `remaining`). See tests for examples. +pub fn last_line(s: &[u8]) -> Option<(&[u8], &[u8])> { + let mut it = memchr::memrchr_iter(b'\n', s); + let last = it.next()?; + let rem = &s[(last + 1)..]; + if let Some(pre_last) = it.next() { + Some((&s[(pre_last + 1)..last], rem)) + } else { + Some((&s[..last], rem)) + } +} + +/// Deserialize the last complete object. Returns (`object`, `remaining`). See tests for examples. +pub fn de_last_json<'a, T: Deserialize<'a>>( + mut s: &'a [u8], +) -> Result<(Option, &'a [u8]), JsonError> { + let mut last = None; + let mut tmp; + loop { + (tmp, s) = de_first_json(s)?; + last = match tmp { + Some(obj) => Some(obj), + None => return Ok((last, s)), + }; + } +} + +/// Deserialize the first complete object. Returns (`object`, `remaining`). See tests for examples. +pub fn de_first_json<'a, T: Deserialize<'a>>( + mut s: &'a [u8], +) -> Result<(Option, &'a [u8]), JsonError> { + while s + .first() + .map_or(false, |&x| x == b' ' || x == b',' || x == b'\n') + { + s = &s[1..]; + } + let mut de = Deserializer::from_slice(s).into_iter(); + match de.next() { + Some(Ok(obj)) => Ok((Some(obj), &s[de.byte_offset()..])), + Some(Err(e)) if e.is_eof() => Ok((None, &s[de.byte_offset()..])), + Some(Err(e)) => Err(e), + None => Ok((None, &s[de.byte_offset()..])), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + macro_rules! str { + ($str:expr) => { + &$str.as_bytes()[..] + }; + } + + #[test] + fn streaming_json() { + let s = b",[2]\n, [3], [4, 3],[32][3] "; + assert_eq!( + de_first_json::>(s).unwrap(), + (Some(vec![2]), str!("\n, [3], [4, 3],[32][3] ")) + ); + assert_eq!( + de_last_json::>(s).unwrap(), + (Some(vec![3]), str!("")) + ); + + let s = b",[2]\n, [3], [4, 3],[32][3] [2, 4"; + assert_eq!( + de_last_json::>(s).unwrap(), + (Some(vec![3]), str!("[2, 4")) + ); + + let s = b",[2]\n, [3], [4, 3],[32] invalid"; + assert_eq!( + de_first_json::>(s).unwrap(), + (Some(vec![2]), str!("\n, [3], [4, 3],[32] invalid")) + ); + assert!(de_last_json::>(s).is_err()); + } + + #[test] + fn test_last_line() { + let s = b"hello"; + assert_eq!(last_line(s), None); + + let s = b"hello\n"; + assert_eq!(last_line(s), Some((str!("hello"), str!("")))); + + let s = b"hello\nworld"; + assert_eq!(last_line(s), Some((str!("hello"), str!("world")))); + + let s = b"hello\nworld\n"; + assert_eq!(last_line(s), Some((str!("world"), str!("")))); + + let s = b"hello\nworld\n..."; + assert_eq!(last_line(s), Some((str!("world"), str!("...")))); + } +} -- cgit v1.3.1