Read epics archiver appliance protobuf file

Cargo.toml

@@ -1,5 +1,5 @@
 [workspace]
-members = ["daqbuffer", "h5out"]
+members = ["daqbuffer", "h5out", "archapp"]
 
 [profile.release]
 debug = 1

archapp/Cargo.toml

@@ -0,0 +1,20 @@
+[package]
+name = "archapp"
+version = "0.0.1-a.dev.4"
+authors = ["Dominik Werder <dominik.werder@gmail.com>"]
+edition = "2018"
+
+[dependencies]
+tokio = { version = "1.4.0", features = ["io-util", "net", "time", "sync", "fs"] }
+tracing = "0.1.25"
+futures-core = "0.3.14"
+futures-util = "0.3.14"
+bytes = "1.0.1"
+serde = "1.0"
+serde_derive = "1.0"
+serde_json = "1.0"
+chrono = "0.4"
+protobuf = "2.24.1"
+err = { path = "../err" }
+taskrun = { path = "../taskrun" }
+netpod = { path = "../netpod" }

archapp/src/generated.rs

@@ -0,0 +1,2 @@
+#[allow(non_snake_case)]
+pub mod EPICSEvent;

archapp/src/lib.rs

@@ -0,0 +1,4 @@
+pub mod generated;
+
+#[cfg(test)]
+mod test;

archapp/src/test.rs

@@ -0,0 +1,63 @@
+use err::Error;
+use netpod::log::*;
+use protobuf::Message;
+
+fn unescape_archapp_msg(inp: &[u8]) -> Result<Vec<u8>, Error> {
+    let mut ret = Vec::with_capacity(inp.len() * 5 / 4);
+    let mut esc = false;
+    for &k in inp.iter() {
+        if k == 0x1b {
+            esc = true;
+        } else if esc {
+            if k == 0x1 {
+                ret.push(0x1b);
+            } else if k == 0x2 {
+                ret.push(0xa);
+            } else if k == 0x3 {
+                ret.push(0xd);
+            } else {
+                return Err(Error::with_msg("malformed escaped archapp message"));
+            }
+            esc = false;
+        } else {
+            ret.push(k);
+        }
+    }
+    Ok(ret)
+}
+
+#[test]
+fn read_pb_00() -> Result<(), Error> {
+    let block1 = async move {
+        let path = "../../../../archappdata/tmp/lts/ArchiverStore/SARUN16/MQUA080/X:2021_01.pb";
+        let f1 = tokio::fs::read(path).await?;
+        let mut j1 = 0;
+        loop {
+            let mut i2 = usize::MAX;
+            for (i1, &k) in f1[j1..].iter().enumerate() {
+                if k == 0xa {
+                    i2 = j1 + i1;
+                    break;
+                }
+            }
+            if i2 != usize::MAX {
+                info!("got NL  {} .. {}", j1, i2);
+                let m = unescape_archapp_msg(&f1[j1..i2])?;
+                if j1 == 0 {
+                    let payload_info = crate::generated::EPICSEvent::PayloadInfo::parse_from_bytes(&m).unwrap();
+                    info!("got payload_info: {:?}", payload_info);
+                } else {
+                    let scalar_double = crate::generated::EPICSEvent::ScalarDouble::parse_from_bytes(&m).unwrap();
+                    info!("got scalar_double: {:?}", scalar_double);
+                }
+            } else {
+                info!("no more packets");
+                break;
+            }
+            j1 = i2 + 1;
+        }
+        Ok::<_, Error>(())
+    };
+    taskrun::run(block1)?;
+    Ok(())
+}