Merge pull request #86 from romand/i32be

Add i32be stream type

Author: sevagh

Cargo.lock

@@ -2,7 +2,7 @@
 
 ### protobuf to json deserializer, written in Rust
 
-`pq` is a tool which deserializes protobuf messages given a set of pre-compiled `.fdset` files. It can understand varint/leb128-delimited streams, and it can connect to Kafka.
+`pq` is a tool which deserializes protobuf messages given a set of pre-compiled `.fdset` files. It can understand varint/leb128-delimited/i32be streams, and it can connect to Kafka.
 
 `pq` will pretty-print when outputting to a tty, but you should pipe it to `jq` for more fully-featured json handling.
 

README.md

@@ -61,6 +61,17 @@ $ pq kafka my_topic --brokers=192.168.0.1:9092 --count 1 --convert varint |\
 }
 ```
 
+Pipe `kafkacat` output to it:
+```
+$ kafkacat -b 192.168.0.1:9092 -C -u -q -f "%R%s" -t my_topic |\
+> pq --msgtype=com.example.dog.Dog --stream i32be
+{
+  "age": 10,
+  "breed": "gsd",
+  "temperament": "aggressive"
+}
+```
+
 ### Compile without kafka
 
 To compile `pq` without kafka support, run:

USAGE.md

@@ -12,6 +12,7 @@ edition = "2018"
 
 [dependencies]
 kafka = { version = "0.7", optional = true }
+byteorder = "1.3"
 
 [features]
 default = []

stream-delimit/Cargo.toml

@@ -1,5 +1,6 @@
 #![deny(missing_docs)]
 
+use crate::i32be::consume_single_i32be;
 use crate::stream::*;
 use crate::varint::consume_single_varint;
 use std::io::Read;
@@ -23,6 +24,7 @@ impl<T: Read> Iterator for ByteConsumer<T> {
     fn next(&mut self) -> Option<Vec<u8>> {
         match self.type_ {
             StreamType::Leb128 | StreamType::Varint => consume_single_varint(&mut self.read),
+            StreamType::I32BE => consume_single_i32be(&mut self.read),
             StreamType::Single => {
                 let ret: Option<Vec<u8>>;
                 let mut buf = Vec::new();

stream-delimit/src/byte_consumer.rs

@@ -0,0 +1,40 @@
+#![deny(missing_docs)]
+
+use byteorder::{BigEndian, ReadBytesExt};
+use std::io::Read;
+
+pub fn consume_single_i32be(read: &mut Read) -> Option<Vec<u8>> {
+    match read.read_i32::<BigEndian>() {
+        Ok(length) => {
+            let mut msg_buf = vec![0; length as usize];
+            match read.read_exact(&mut msg_buf) {
+                Ok(_) => Some(msg_buf),
+                Err(_) => None,
+            }
+        }
+        Err(_) => None,
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use byteorder::WriteBytesExt;
+    use std::io::Cursor;
+
+    fn reads_back(x: i32) {
+        let mut buf = vec![];
+        buf.write_i32::<BigEndian>(x).unwrap();
+        assert_eq!(x, Cursor::new(buf).read_i32::<BigEndian>().unwrap());
+    }
+
+    #[test]
+    fn test_simple() {
+        reads_back(1);
+    }
+
+    #[test]
+    fn test_delimiter_longer() {
+        reads_back(300);
+    }
+}

stream-delimit/src/i32be.rs

@@ -1,6 +1,8 @@
+extern crate byteorder;
 #[cfg(feature = "with_kafka")]
 extern crate kafka;
 
+mod i32be;
 mod varint;
 
 pub mod error;

stream-delimit/src/lib.rs

@@ -8,6 +8,9 @@ pub enum StreamType {
     Leb128,
     /// Protobuf messages with leading length encoded in varint
     Varint,
+    /// Protobuf messages with leading length encoded as
+    /// binary big endian 32-bit signed integer
+    I32BE,
     /// Single protobuf messages with no separators/delimiters
     Single,
 }
@@ -18,6 +21,7 @@ pub fn str_to_streamtype(input: &str) -> Result<StreamType> {
         "single" => Ok(StreamType::Single),
         "varint" => Ok(StreamType::Varint),
         "leb128" => Ok(StreamType::Leb128),
+        "i32be" => Ok(StreamType::I32BE),
         _ => Err(StreamDelimitError::InvalidStreamTypeError(
             input.to_string(),
         ))?,

stream-delimit/src/stream.rs

@@ -35,6 +35,19 @@ fn test_dog_decode_stream() {
         .unwrap();
 }
 
+#[test]
+fn test_dog_decode_i32be_stream() {
+    assert_cli::Assert::main_binary()
+        .with_env(assert_cli::Environment::inherit().insert("FDSET_PATH", get_fdset_dir("fdsets")))
+        .with_args(&["--msgtype=com.example.dog.Dog", "--stream=i32be"])
+        .stdin(include_str!("samples/dog_i32be_stream"))
+        .succeeds()
+        .and()
+        .stdout()
+        .contains("{\"age\":3,\"breed\":\"gsd\",\"temperament\":\"excited\"}")
+        .unwrap();
+}
+
 #[test]
 fn test_nonexistent_fdset_dir() {
     assert_cli::Assert::main_binary()