test the Streaming wrapper on HTTP benchmarks

benchmarks/benches/http_streaming.rs

@@ -4,7 +4,7 @@
 static ALLOC: jemallocator::Jemalloc = jemallocator::Jemalloc;
 
 use criterion::*;
-use nom::{IResult, bytes::streaming::{tag, take_while1}, character::streaming::{line_ending, char}, multi::many};
+use nom::{IResult, bytes::{tag,  take_while1}, character::{line_ending, char}, multi::many, Streaming};
 
 #[cfg_attr(rustfmt, rustfmt_skip)]
 #[derive(Debug)]
@@ -67,41 +67,41 @@ fn is_version(c: u8) -> bool {
   c >= b'0' && c <= b'9' || c == b'.'
 }
 
-fn request_line(input: &[u8]) -> IResult<&[u8], Request<'_>> {
+fn request_line(input: Streaming<&[u8]>) -> IResult<Streaming<&[u8]>, Request<'_>> {
   let (input, method) = take_while1(is_token)(input)?;
   let (input, _) = take_while1(is_space)(input)?;
   let (input, uri) = take_while1(is_not_space)(input)?;
   let (input, _) = take_while1(is_space)(input)?;
   let (input, version) = http_version(input)?;
   let (input, _) = line_ending(input)?;
 
-  Ok((input, Request {method, uri, version}))
+  Ok((input, Request {method: method.into_inner(), uri: uri.into_inner(), version}))
 }
 
-fn http_version(input: &[u8]) -> IResult<&[u8], &[u8]> {
+fn http_version(input: Streaming<&[u8]>) -> IResult<Streaming<&[u8]>, &[u8]> {
   let (input, _) = tag("HTTP/")(input)?;
   let (input, version) = take_while1(is_version)(input)?;
 
-  Ok((input, version))
+  Ok((input, version.into_inner()))
 }
 
-fn message_header_value(input: &[u8]) -> IResult<&[u8], &[u8]> {
+fn message_header_value(input: Streaming<&[u8]>) -> IResult<Streaming<&[u8]>, &[u8]> {
   let (input, _) = take_while1(is_horizontal_space)(input)?;
   let (input, data) = take_while1(not_line_ending)(input)?;
   let (input, _) = line_ending(input)?;
 
-  Ok((input, data))
+  Ok((input, data.into_inner()))
 }
 
-fn message_header(input: &[u8]) -> IResult<&[u8], Header<'_>> {
+fn message_header(input: Streaming<&[u8]>) -> IResult<Streaming<&[u8]>, Header<'_>> {
   let (input, name) = take_while1(is_token)(input)?;
   let (input, _) = char(':')(input)?;
   let (input, value) = many(1.., message_header_value)(input)?;
 
-  Ok((input, Header{ name, value }))
+  Ok((input, Header{ name: name.into_inner(), value }))
 }
 
-fn request(input: &[u8]) -> IResult<&[u8], (Request<'_>, Vec<Header<'_>>)> {
+fn request(input: Streaming<&[u8]>) -> IResult<Streaming<&[u8]>, (Request<'_>, Vec<Header<'_>>)> {
   let (input, req) = request_line(input)?;
   let (input, h) = many(1.., message_header)(input)?;
   let (input, _) = line_ending(input)?;
@@ -114,8 +114,9 @@ fn parse(data: &[u8]) -> Option<Vec<(Request<'_>, Vec<Header<'_>>)>> {
   let mut buf = &data[..];
   let mut v = Vec::new();
   loop {
-    match request(buf) {
+    match request(Streaming::new(buf)) {
       Ok((b, r)) => {
+        let b = b.into_inner();
         buf = b;
         v.push(r);
 

src/bytes/mod.rs

@@ -1,6 +1,91 @@
 //! Parsers recognizing bytes streams
 
+use crate::{
+  error::{ErrorKind, ParseError},
+  Compare, CompareResult, Err, IResult, Input, InputLength, Needed,
+};
+
 pub mod complete;
 pub mod streaming;
 #[cfg(test)]
 mod tests;
+
+/// Recognizes a pattern.
+///
+/// The input data will be compared to the tag combinator's argument and will return the part of
+/// the input that matches the argument.
+/// # Example
+/// ```rust
+/// # use nom::{Err, error::{Error, ErrorKind}, Needed, IResult};
+/// use nom::bytes::streaming::tag;
+///
+/// fn parser(s: &str) -> IResult<&str, &str> {
+///   tag("Hello")(s)
+/// }
+///
+/// assert_eq!(parser("Hello, World!"), Ok((", World!", "Hello")));
+/// assert_eq!(parser("Something"), Err(Err::Error(Error::new("Something", ErrorKind::Tag))));
+/// assert_eq!(parser("S"), Err(Err::Error(Error::new("S", ErrorKind::Tag))));
+/// assert_eq!(parser("H"), Err(Err::Incomplete(Needed::new(4))));
+/// ```
+pub fn tag<T, I, Error: ParseError<I>>(tag: T) -> impl Fn(I) -> IResult<I, I, Error>
+where
+  I: Input + Compare<T>,
+  T: InputLength + Clone,
+{
+  move |i: I| {
+    let tag_len = tag.input_len();
+    let t = tag.clone();
+
+    let res: IResult<_, _, Error> = match i.compare(t) {
+      CompareResult::Ok => Ok(i.take_split(tag_len)),
+      CompareResult::Incomplete if I::is_streaming() => {
+        Err(Err::Incomplete(Needed::new(tag_len - i.input_len())))
+      }
+      _ => {
+        let e: ErrorKind = ErrorKind::Tag;
+        Err(Err::Error(Error::from_error_kind(i, e)))
+      }
+    };
+    res
+  }
+}
+
+/// Returns the longest (at least 1) input slice that matches the predicate.
+///
+/// The parser will return the longest slice that matches the given predicate *(a function that
+/// takes the input and returns a bool)*.
+///
+/// It will return an `Err(Err::Error((_, ErrorKind::TakeWhile1)))` if the pattern wasn't met.
+///
+/// # Streaming Specific
+/// *Streaming version* will return a `Err::Incomplete(Needed::new(1))` or if the pattern reaches the end of the input.
+///
+/// # Example
+/// ```rust
+/// # use nom::{Err, error::{Error, ErrorKind}, Needed, IResult};
+/// use nom::bytes::streaming::take_while1;
+/// use nom::character::is_alphabetic;
+///
+/// fn alpha(s: &[u8]) -> IResult<&[u8], &[u8]> {
+///   take_while1(is_alphabetic)(s)
+/// }
+///
+/// assert_eq!(alpha(b"latin123"), Ok((&b"123"[..], &b"latin"[..])));
+/// assert_eq!(alpha(b"latin"), Err(Err::Incomplete(Needed::new(1))));
+/// assert_eq!(alpha(b"12345"), Err(Err::Error(Error::new(&b"12345"[..], ErrorKind::TakeWhile1))));
+/// ```
+pub fn take_while1<F, I, Error: ParseError<I>>(cond: F) -> impl Fn(I) -> IResult<I, I, Error>
+where
+  I: Input,
+  F: Fn(<I as Input>::Item) -> bool,
+{
+  move |i: I| {
+    let e: ErrorKind = ErrorKind::TakeWhile1;
+    if I::is_streaming() {
+      i.split_at_position1(|c| !cond(c), e)
+    } else {
+      i.split_at_position1_complete(|c| !cond(c), e)
+    }
+  }
+}

src/character/mod.rs

@@ -2,6 +2,11 @@
 //!
 //! Functions recognizing specific characters
 
+use crate::{
+  error::{ErrorKind, ParseError},
+  AsChar, Compare, CompareResult, Err, IResult, Input, Needed,
+};
+
 #[cfg(test)]
 mod tests;
 
@@ -114,3 +119,82 @@ pub fn is_space(chr: u8) -> bool {
 pub fn is_newline(chr: u8) -> bool {
   chr == b'\n'
 }
+
+/// Recognizes one character.
+///
+/// *Streaming version*: Will return `Err(nom::Err::Incomplete(_))` if there's not enough input data.
+/// # Example
+///
+/// ```
+/// # use nom::{Err, error::{ErrorKind, Error}, Needed, IResult};
+/// # use nom::character::streaming::char;
+/// fn parser(i: &str) -> IResult<&str, char> {
+///     char('a')(i)
+/// }
+/// assert_eq!(parser("abc"), Ok(("bc", 'a')));
+/// assert_eq!(parser("bc"), Err(Err::Error(Error::new("bc", ErrorKind::Char))));
+/// assert_eq!(parser(""), Err(Err::Incomplete(Needed::new(1))));
+/// ```
+pub fn char<I, Error: ParseError<I>>(c: char) -> impl Fn(I) -> IResult<I, char, Error>
+where
+  I: Input,
+  <I as Input>::Item: AsChar,
+{
+  move |i: I| match (i).iter_elements().next().map(|t| {
+    let b = t.as_char() == c;
+    (&c, b)
+  }) {
+    None => {
+      if I::is_streaming() {
+        Err(Err::Incomplete(Needed::new(c.len() - i.input_len())))
+      } else {
+        Err(Err::Error(Error::from_char(i, c)))
+      }
+    }
+    Some((_, false)) => Err(Err::Error(Error::from_char(i, c))),
+    Some((c, true)) => Ok((i.take_from(c.len()), c.as_char())),
+  }
+}
+
+/// Recognizes an end of line (both '\n' and '\r\n').
+///
+/// *Streaming version*: Will return `Err(nom::Err::Incomplete(_))` if there's not enough input data.
+/// # Example
+///
+/// ```
+/// # use nom::{Err, error::ErrorKind, IResult, Needed};
+/// # use nom::character::streaming::line_ending;
+/// assert_eq!(line_ending::<_, (_, ErrorKind)>("\r\nc"), Ok(("c", "\r\n")));
+/// assert_eq!(line_ending::<_, (_, ErrorKind)>("ab\r\nc"), Err(Err::Error(("ab\r\nc", ErrorKind::CrLf))));
+/// assert_eq!(line_ending::<_, (_, ErrorKind)>(""), Err(Err::Incomplete(Needed::new(1))));
+/// ```
+pub fn line_ending<I, E: ParseError<I>>(input: I) -> IResult<I, I, E>
+where
+  I: Input,
+  I: Compare<&'static str>,
+{
+  match input.compare("\n") {
+    CompareResult::Ok => Ok(input.take_split(1)),
+    CompareResult::Incomplete => {
+      if I::is_streaming() {
+        Err(Err::Incomplete(Needed::new(1)))
+      } else {
+        Err(Err::Error(E::from_error_kind(input, ErrorKind::CrLf)))
+      }
+    }
+    CompareResult::Error => {
+      match input.compare("\r\n") {
+        //FIXME: is this the right index?
+        CompareResult::Ok => Ok(input.take_split(2)),
+        CompareResult::Incomplete => {
+          if I::is_streaming() {
+            Err(Err::Incomplete(Needed::new(2)))
+          } else {
+            Err(Err::Error(E::from_error_kind(input, ErrorKind::CrLf)))
+          }
+        }
+        CompareResult::Error => Err(Err::Error(E::from_error_kind(input, ErrorKind::CrLf))),
+      }
+    }
+  }
+}

src/multi/mod.rs

@@ -1080,7 +1080,7 @@ where
 #[cfg_attr(feature = "docsrs", doc(cfg(feature = "alloc")))]
 pub fn many<I, O, E, F, G>(range: G, mut parse: F) -> impl FnMut(I) -> IResult<I, Vec<O>, E>
 where
-  I: Clone + InputLength,
+  I: Clone + Input,
   F: Parser<I, O, E>,
   E: ParseError<I>,
   G: NomRange<usize>,