Make our own LinesCodec

src/commands/classified.rs

@@ -1,7 +1,46 @@
 use crate::prelude::*;
-use futures_codec::{Framed, LinesCodec};
+use futures::TryStreamExt;
+use futures_codec::{Encoder, Decoder, Framed};
 use std::sync::Arc;
 use subprocess::Exec;
+use std::io::{Error, ErrorKind};
+use bytes::{BufMut, BytesMut};
+
+/// A simple `Codec` implementation that splits up data into lines.
+pub struct LinesCodec {}
+
+impl Encoder for LinesCodec {
+    type Item = String;
+    type Error = Error;
+
+    fn encode(&mut self, item: Self::Item, dst: &mut BytesMut) -> Result<(), Self::Error> {
+        dst.put(item);
+        Ok(())
+    }
+}
+
+impl Decoder for LinesCodec {
+    type Item = String;
+    type Error = Error;
+
+    fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
+        match src.iter().position(|b| b == &b'\n') {
+            Some(pos) if !src.is_empty() => {
+                let buf = src.split_to(pos + 1);
+                String::from_utf8(buf.to_vec())
+                    .map(Some)
+                    .map_err(|e| Error::new(ErrorKind::InvalidData, e))
+            }
+            _ if !src.is_empty() => {
+                let drained = src.take();
+                String::from_utf8(drained.to_vec())
+                    .map(Some)
+                    .map_err(|e| Error::new(ErrorKind::InvalidData, e))
+            }
+            _ => Ok(None)
+        }
+    }
+}
 
 crate struct ClassifiedInputStream {
     crate objects: InputStream,