/// decode stream to header and message
decode stream to header and message
decode stream to header and message
pub async fn read_msg<T: AsyncRead + Unpin + Send>(
    buf_reader: &mut BufReader<T>,
) -> Result<(Header, MongoMsg)> {
    let header = Header::read_from(buf_reader).await?;
    // TODO: RUST-616 ensure length is < maxMessageSizeBytes
    let length_remaining = header.length - Header::LENGTH as i32;
    let mut buf = vec![0u8; length_remaining as usize];
    buf_reader.read_exact(&mut buf).await?;
    let reader = buf.as_slice();
    let msg = if header.op_code == OpCode::Message {
        MongoMsg::Message(Message::read_op_common(reader, length_remaining, &header)?)
    } else if header.op_code == OpCode::Query {
        MongoMsg::Query(QueryOp::read_query_common(reader, length_remaining, &header).await?)
    } else if header.op_code == OpCode::Reply {
        MongoMsg::Reply(ReplyOp::read_reply_common(reader, length_remaining, &header).await?)
        //} else if header.op_code == OpCode::Compressed {
        //    MongoMsg::Message(Message::read_from_op_compressed(reader, &header).await?)
    } else {
        return Err(Error::new(
            ErrorKind::InvalidResponse {
                message: format!(
                    "Invalid op code, expected {} or {} and got {}",
                    OpCode::Message as u32,
                    OpCode::Compressed as u32,
                    header.op_code as u32
                ),
            },
            Option::<Vec<String>>::None,
        ));
    };
    Ok((header, msg))
}
pub async fn read_msg < T : AsyncRead + Unpin + Send > ( buf_reader : & mut BufReader < T > , ) -> Result < ( Header , MongoMsg ) > { let header = Header :: read_from ( buf_reader ) . await ? ; // TODO: RUST-616 ensure length is < maxMessageSizeBytes let length_remaining = header . length - Header :: LENGTH as i32 ; let mut buf = vec ! [ 0u8 length_remaining as usize ] ; buf_reader . read_exact ( & mut buf ) . await ? ; let reader = buf . as_slice ( ) ; let msg = if header . op_code == OpCode :: Message { MongoMsg :: Message ( Message :: read_op_common ( reader , length_remaining , & header ) ? ) } else if header . op_code == OpCode :: Query { MongoMsg :: Query ( QueryOp :: read_query_common ( reader , length_remaining , & header ) . await ? ) } else if header . op_code == OpCode :: Reply { MongoMsg :: Reply ( ReplyOp :: read_reply_common ( reader , length_remaining , & header ) . await ? ) //} else if header.op_code == OpCode::Compressed { //    MongoMsg::Message(Message::read_from_op_compressed(reader, &header).await?) } else { return Err ( Error :: new ( ErrorKind :: InvalidResponse { message : format ! ( " " OpCode Message as u32 OpCode Compressed as u32 header op_code as u32 ) , } , Option :: < Vec < String > > :: None , ) ) ; } ; Ok ( ( header , msg ) ) }
decode stream to header and message
decode stream to header and message
buf_reader
&mut BufReader<T>