streams3.rs

struct Parser {
    trailers: Option<Receiver<(String, String)>>,
}

impl Parser {
    pub fn new() -> Self {
        Self { trailers: None }
    }

    pub fn parse<'a, T: async_std::io::Read + Unpin + Send + 'static>(
        &mut self,
        mut incoming_bytes: BlockBuffer<T>,
    ) -> AsyncReader<Receiver<Block<Vec<u8>>>> {
        let (sender_data, reader_data) = channel(10);
        let (sender_trailer, reader_trailer) = channel(10);

        self.trailers = Some(reader_trailer);
        async_std::task::spawn(async move {
            loop {
                // read the length of the chunk
                let hex_block: Block = incoming_bytes.read_until(b"\r\n").await.unwrap();
                let len = parse_chunk_size(&hex_block[..]);
                if len == 0 {
                    // we are done with the main parts
                    break;
                }
                // read the actual data
                let data: Block = incoming_bytes.take(len).await.unwrap();
                sender_data.send(data).await;

                // read the trailing \r\n
                let crlf = incoming_bytes.take(2).await.unwrap();
                assert_eq!(&crlf[..], "\r\n".as_bytes());
            }

            // parse trailers
            let mut trailers = incoming_bytes
                .split_at(b"\r\n") // create chunks splitting at each crlf boundary
                .map(|chunk: Result<Block>| {
                    let chunk = chunk.unwrap();
                    // split into the first occururence of ": "
                    let i = bytes!(':', ' ').find(&chunk[..]).unwrap();
                    let (key, value) = chunk.split_at(i);
                    (
                        String::from_utf8(key.into()).unwrap(),
                        String::from_utf8(value[2..].into()).unwrap(),
                    )
                });

            while let Some(trailer) = trailers.next().await {
                sender_trailer.send(trailer).await;
            }
        });

        AsyncReader::new(reader_data)
    }
}

let mut parser = Parser::new();

let incoming = Cursor::new(
    "7\r\n\
 Mozilla\r\n\
 9\r\n\
 Developer\r\n\
 7\r\n\
 Network\r\n\
 0\r\n\
 Expires: Wed, 21 Oct 2015 07:28:00 GMT\r\n\
 \r\n"
        .as_bytes(),
);

let incoming_bytes = BlockBuffer::new(pool.clone(), incoming);
let mut s = parser.parse(incoming_bytes);

let mut result = Vec::new();
s.read_to_end(&mut result).await.unwrap();

let trailers = parser.trailers.unwrap().collect::<Vec<_>>().await;

assert_eq!(result, "MozillaDeveloperNetwork".as_bytes());
assert_eq!(
    trailers,
    vec![(
        "Expires".to_string(),
        "Wed, 21 Oct 2015 07:28:00 GMT".to_string()
    )]
);