Refactor and add test for api1 binary query

daqbufp2/Cargo.toml

@@ -4,6 +4,9 @@ version = "0.0.1-a.dev.12"
 authors = ["Dominik Werder <dominik.werder@gmail.com>"]
 edition = "2021"
 
+[lib]
+path = "src/daqbufp2.rs"
+
 [dependencies]
 tokio = { version = "1.22.0", features = ["rt-multi-thread", "io-util", "net", "time", "sync", "fs"] }
 hyper = "0.14"
@@ -27,3 +30,6 @@ httpclient = { path = "../httpclient" }
 disk = { path = "../disk" }
 items = { path = "../items" }
 streams = { path = "../streams" }
+
+[dev-dependencies]
+nom = "7.1.1"

daqbufp2/src/daqbufp2.rs

@@ -21,6 +21,9 @@ pub fn spawn_test_hosts(cluster: Cluster) -> Vec<JoinHandle<Result<(), Error>>>
         let h = tokio::spawn(httpret::host(node_config).map_err(Error::from));
         ret.push(h);
     }
+
+    // TODO spawn also two proxy nodes
+
     ret
 }
 

daqbufp2/src/test/api1.rs

@@ -1,13 +1,15 @@
-use crate::err::ErrConv;
+#[cfg(test)]
+mod api1_parse;
+
 use crate::nodes::require_test_hosts_running;
-use chrono::{DateTime, Utc};
+use crate::test::api1::api1_parse::Api1Frame;
 use err::Error;
 use futures_util::Future;
-use http::{header, Request, StatusCode};
-use httpclient::{http_get, http_post};
-use hyper::Body;
+use httpclient::http_post;
+use httpret::api1::Api1ScalarType;
 use netpod::log::*;
 use netpod::query::api1::{Api1Query, Api1Range, ChannelTuple};
+use std::fmt;
 use url::Url;
 
 fn testrun<T, F>(fut: F) -> Result<T, Error>
@@ -17,6 +19,29 @@ where
     taskrun::run(fut)
 }
 
+fn is_monitonic_strict<I>(it: I) -> bool
+where
+    I: Iterator,
+    <I as Iterator>::Item: PartialOrd + fmt::Debug,
+{
+    let mut last = None;
+    for x in it {
+        if let Some(last) = last.as_ref() {
+            if x <= *last {
+                return false;
+            }
+        }
+        last = Some(x);
+    }
+    true
+}
+
+#[test]
+fn test_is_monitonic_strict() {
+    assert_eq!(is_monitonic_strict([1, 2, 3].iter()), true);
+    assert_eq!(is_monitonic_strict([1, 2, 2].iter()), false);
+}
+
 #[test]
 fn events_f64_plain() -> Result<(), Error> {
     let fut = async {
@@ -32,8 +57,55 @@ fn events_f64_plain() -> Result<(), Error> {
         let body = serde_json::to_string(&qu)?;
         let buf = http_post(url, accept, body.into()).await?;
         eprintln!("body received: {}", buf.len());
-        //let js = String::from_utf8_lossy(&buf);
-        //eprintln!("string received: {js}");
+        match api1_parse::api1_frames(&buf) {
+            Ok((_, frames)) => {
+                debug!("FRAMES LEN: {}", frames.len());
+                assert_eq!(frames.len(), 121);
+                if let Api1Frame::Header(header) = &frames[0] {
+                    if let Api1ScalarType::I32 = header.header().ty() {
+                    } else {
+                        panic!("bad scalar type")
+                    }
+                } else {
+                    panic!("bad header")
+                }
+                let tss: Vec<_> = frames
+                    .iter()
+                    .filter_map(|f| match f {
+                        api1_parse::Api1Frame::Data(d) => Some(d.ts()),
+                        _ => None,
+                    })
+                    .collect();
+                let pulses: Vec<_> = frames
+                    .iter()
+                    .filter_map(|f| match f {
+                        api1_parse::Api1Frame::Data(d) => Some(d.pulse()),
+                        _ => None,
+                    })
+                    .collect();
+                let values: Vec<_> = frames
+                    .iter()
+                    .filter_map(|f| match f {
+                        api1_parse::Api1Frame::Data(d) => {
+                            let val = i32::from_be_bytes(d.data().try_into().unwrap());
+                            Some(val)
+                        }
+                        _ => None,
+                    })
+                    .collect();
+                assert_eq!(is_monitonic_strict(tss.iter()), true);
+                assert_eq!(is_monitonic_strict(pulses.iter()), true);
+                assert_eq!(is_monitonic_strict(values.iter()), true);
+                for &val in &values {
+                    assert!(val >= 0);
+                    assert!(val < 120);
+                }
+            }
+            Err(e) => {
+                error!("can not parse result: {e}");
+                panic!()
+            }
+        };
         Ok(())
     };
     testrun(fut)?;

daqbufp2/src/test/api1/api1_parse.rs

@@ -0,0 +1,184 @@
+use httpret::api1::Api1ChannelHeader;
+use netpod::log::*;
+use nom::multi::many0;
+use nom::number::complete::{be_u32, be_u64, be_u8};
+use nom::IResult;
+use std::num::NonZeroUsize;
+
+// u32be length_1.
+// there is exactly length_1 more bytes in this message.
+// u8 mtype: 0: channel-header, 1: data
+
+// for mtype == 0:
+// The rest is a JSON with the channel header.
+
+// for mtype == 1:
+// u64be timestamp
+// u64be pulse
+// After that comes exactly (length_1 - 17) bytes of data.
+
+#[derive(Debug)]
+pub struct Header {
+    header: Api1ChannelHeader,
+}
+
+impl Header {
+    pub fn header(&self) -> &Api1ChannelHeader {
+        &self.header
+    }
+}
+
+#[derive(Debug)]
+pub struct Data {
+    ts: u64,
+    pulse: u64,
+    data: Vec<u8>,
+}
+
+impl Data {
+    pub fn ts(&self) -> u64 {
+        self.ts
+    }
+
+    pub fn pulse(&self) -> u64 {
+        self.pulse
+    }
+
+    pub fn data(&self) -> &[u8] {
+        &self.data
+    }
+}
+
+#[derive(Debug)]
+pub enum Api1Frame {
+    Header(Header),
+    Data(Data),
+}
+
+fn header(inp: &[u8]) -> IResult<&[u8], Header> {
+    match serde_json::from_slice(inp) {
+        Ok(k) => {
+            let k: Api1ChannelHeader = k;
+            IResult::Ok((&inp[inp.len()..], Header { header: k }))
+        }
+        Err(e) => {
+            error!("json header parse error: {e}");
+            let e = nom::Err::Failure(nom::error::make_error(inp, nom::error::ErrorKind::Fail));
+            IResult::Err(e)
+        }
+    }
+}
+
+fn data(inp: &[u8]) -> IResult<&[u8], Data> {
+    if inp.len() < 16 {
+        use nom::{Err, Needed};
+        return IResult::Err(Err::Incomplete(Needed::Size(NonZeroUsize::new(16).unwrap())));
+    }
+    let (inp, ts) = be_u64(inp)?;
+    let (inp, pulse) = be_u64(inp)?;
+    let data = inp.into();
+    let inp = &inp[inp.len()..];
+    let res = Data { ts, pulse, data };
+    IResult::Ok((inp, res))
+}
+
+fn api1_frame_complete(inp: &[u8]) -> IResult<&[u8], Api1Frame> {
+    let (inp, mtype) = be_u8(inp)?;
+    if mtype == 0 {
+        let (inp, val) = header(inp)?;
+        if inp.len() != 0 {
+            // We did not consume the exact number of bytes
+            let kind = nom::error::ErrorKind::Verify;
+            let e = nom::error::Error::new(inp, kind);
+            Err(nom::Err::Failure(e))
+        } else {
+            let res = Api1Frame::Header(val);
+            IResult::Ok((inp, res))
+        }
+    } else if mtype == 1 {
+        let (inp, val) = data(inp)?;
+        if inp.len() != 0 {
+            // We did not consume the exact number of bytes
+            let kind = nom::error::ErrorKind::Verify;
+            let e = nom::error::Error::new(inp, kind);
+            Err(nom::Err::Failure(e))
+        } else {
+            let res = Api1Frame::Data(val);
+            IResult::Ok((inp, res))
+        }
+    } else {
+        let e = nom::Err::Incomplete(nom::Needed::Size(NonZeroUsize::new(1).unwrap()));
+        IResult::Err(e)
+    }
+}
+
+fn api1_frame(inp: &[u8]) -> IResult<&[u8], Api1Frame> {
+    let (inp, len) = be_u32(inp)?;
+    if len < 1 {
+        use nom::error::{ErrorKind, ParseError};
+        use nom::Err;
+        return IResult::Err(Err::Failure(ParseError::from_error_kind(inp, ErrorKind::Fail)));
+    }
+    if inp.len() < len as usize {
+        let e = nom::Err::Incomplete(nom::Needed::Size(NonZeroUsize::new(len as _).unwrap()));
+        IResult::Err(e)
+    } else {
+        let (inp2, inp) = inp.split_at(len as _);
+        let (inp2, res) = api1_frame_complete(inp2)?;
+        if inp2.len() != 0 {
+            let kind = nom::error::ErrorKind::Fail;
+            let e = nom::error::Error::new(inp, kind);
+            IResult::Err(nom::Err::Failure(e))
+        } else {
+            IResult::Ok((inp, res))
+        }
+    }
+}
+
+type Nres<'a, T> = IResult<&'a [u8], T, nom::error::VerboseError<&'a [u8]>>;
+
+#[allow(unused)]
+fn verbose_err(inp: &[u8]) -> Nres<u32> {
+    use nom::error::{ErrorKind, ParseError, VerboseError};
+    use nom::Err;
+    let e = VerboseError::from_error_kind(inp, ErrorKind::Fail);
+    return IResult::Err(Err::Failure(e));
+}
+
+pub fn api1_frames(inp: &[u8]) -> IResult<&[u8], Vec<Api1Frame>> {
+    let (inp, res) = many0(api1_frame)(inp)?;
+    IResult::Ok((inp, res))
+}
+
+#[test]
+fn test_basic_frames() -> Result<(), err::Error> {
+    use std::io::Write;
+    let mut buf = Vec::new();
+    let js = r#"{"name": "ch1", "type": "float64", "byteOrder": "LITTLE_ENDIAN"}"#;
+    buf.write(&(1 + js.as_bytes().len() as u32).to_be_bytes())?;
+    buf.write(&[0])?;
+    buf.write(js.as_bytes())?;
+
+    buf.write(&25u32.to_be_bytes())?;
+    buf.write(&[1])?;
+    buf.write(&20u64.to_be_bytes())?;
+    buf.write(&21u64.to_be_bytes())?;
+    buf.write(&5.123f64.to_be_bytes())?;
+
+    buf.write(&25u32.to_be_bytes())?;
+    buf.write(&[1])?;
+    buf.write(&22u64.to_be_bytes())?;
+    buf.write(&23u64.to_be_bytes())?;
+    buf.write(&7.88f64.to_be_bytes())?;
+
+    match api1_frames(&buf) {
+        Ok((_, frames)) => {
+            assert_eq!(frames.len(), 3);
+        }
+        Err(e) => {
+            error!("can not parse result: {e}");
+            panic!()
+        }
+    };
+    Ok(())
+}