1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
use crate::{Decoder, Encoder};
use bytes::{BufMut, BytesMut};
use std::io::{Error, ErrorKind};
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))
}
_ => Ok(None),
}
}
}
#[cfg(test)]
mod test {
use super::*;
use crate::FramedRead;
use futures::{executor, TryStreamExt};
use std::io::Cursor;
#[test]
fn it_works() {
let buf = "Hello\nWorld\nError".to_owned();
let cur = Cursor::new(buf);
let mut framed = FramedRead::new(cur, LinesCodec {});
let next = executor::block_on(framed.try_next()).unwrap().unwrap();
assert_eq!(next, "Hello\n");
let next = executor::block_on(framed.try_next()).unwrap().unwrap();
assert_eq!(next, "World\n");
assert!(executor::block_on(framed.try_next()).is_err());
}
}