Como ler leb128 de TcpStream em Rust

Não tenho certeza, mas acho que ele é bloqueado no let file_name_len = leb128::read::unsigned(&mut socket)?;, o que de alguma forma faz com que meu ouvinte TCP pare de aceitar novas conexões, o que é ainda mais estranho.

O código acima está bloqueando :

1. Você define o soquete para bloqueio ( set_nonblocking(false)).
2. Então bloqueie em leb128::read::unsigned(&mut socket)?;.

Isso bloqueará todo o tópico tokio.

Ele não deve bloquear o ouvinte TCP se o ouvinte TCP for executado em uma tarefa separada e você estiver usando o tempo de execução multithread (padrão) do Tokio... a menos que, é claro, você tenha várias tarefas LEB bloqueando cada thread do Tokio.

Infelizmente, não há uma API padrão para leituras assíncronas, e o leb128pacote não fornece nenhuma integração com o Tokio, então vai dar um pouco de trabalho.

Não muito, porém, porque &[u8]implementa Read, e depois Reado slice terá sido atualizado para apontar para os bytes não lidos.

Como você está usando TCP, presumo que já tenha algum tipo de buffer para os bytes recebidos — para passá-los ao decodificador — então você deve usar apenas esse buffer .

//  Read until sufficient bytes are obtained to determine length.
let length = loop {
    const UNEXPECTED_EOF: io::ErrorKind = io::ErrorKind::UnexpectedEof;

    let mut slice = buffer.readable();

    let result = leb128::read::unsigned(&mut slice);

    match result {
        Ok(length) => {
            let consumed = buffer.readable().len() - slice.len();

            buffer.advance(consumed);

            break length;
        },
        Err(leb128::read::Error::IoError(e)) if e.kind() == UNEXPECTED_EOF =>
            continue,
        Err(e) => return Err(e.into()),
    }

    socket.readable().await?;

    let length = socket.try_read(buffer.writable())?;

    todo!("Handle length, beware 0 means EOF");
};

//  Do something with buffered bytes, perhaps waiting for more
//  (now that you know how many you need).
todo!();

Não acho a estrutura de código acima muito... legal, no entanto. Misturar E/S assíncrona e decodificação significa que você não pode testar a decodificação sozinha, doloroso, eu realmente aconselho preferir o design Sans IO quando possível.

Em vez disso, eu o encorajaria a escrever um Framerou Decoderque cuidará de parte (ou toda) da lógica de decodificação e apenas separará claramente a E/S do enquadramento/decodificação.

A ideia é relativamente simples: inserir bytes nele e obter bytes enquadrados ou mensagens decodificadas.

Como não tenho seu decodificador, usarei um framer, cuja função é isolar um único quadro (mensagem codificada) no fluxo.

Depois que você tiver um emoldurador, na verdade é relativamente simples:

socket.readable().await?;

let mut buffer = [0; 16 * 1024];
let length = socket.try_read(&mut buffer)?;

if length == 0 {
    todo!("handle EOF");
}

framer.push(&buffer[..length]);

//  May want to limit the number of frames handled at once, to avoid blocking
//  other clients.

while let Some(message) = framer.pop()? {
    todo!("handle message");
}

E o mais importante: é muito fácil testar se o criador da mensagem consegue lidar com todos os tipos de mensagens.

O código do framer real é relativamente simples:

//  Disclaimer: I have not even _compiled_ this code, don't expect it to handle
//              all the edge cases.

#[derive(Clone, Debug, Default)]
pub struct Framer {
    state: FramerState,
    consumed: usize,
    buffer: Vec<u8>,
}

impl Framer {
    /// Constructs a framer with a specific capacity.
    pub fn with_capacity(capacity: usize) -> Self {
        let state = FramerState::default();
        let consumed = 0;
        let buffer = Vec::with_capacity(capacity);

        Self { state, consumed, buffer }
    }

    /// Pushes bytes into the framer.
    pub fn push(&mut self, bytes: &[u8]) {
        //  Trick here: draining in push is easier, and avoids O(N²) pop.
        if self.consumed > 0 {
            self.buffer.drain(..self.consumed);
        }

        self.buffer.extend_from_slice(bytes);
    }

    /// Pops a frame, if possible.
    ///
    /// Call repeatedly to pop all buffered frames, when no complete frame is
    /// buffered any longer, returns `Ok(None)`.
    ///
    /// Returns an error if the underlying stream is faulty, for example has
    /// an overflowing length.
    pub fn pop(&mut self) -> Result<Option<&[u8]>, Error> {
        match self.state {
            FramerState::WaitingForLength => {
                let length = self.pop_length()?;

                let Some(length) = length else { return Ok(None) };

                let Some(length) = NonZeroU64::new(length) else {
                    return Ok(Some(&[]));
                };

                //  FIXME: may want a maximum length here, regardless of
                //         overflow, as otherwise a client accidentally
                //         sending 2^63-1 LEB encoded will lead the server
                //         to wait forever.

                self.state = FramerState::WaitingForMessage(length);

                self.pop_message(length)
            }
            FramerState::WaitingForMessage(length) => {
                self.pop_message(length)
            }
        }
    }

    //  Pops the length.
    //
    //  # Panics
    //
    //  In Debug, if called when state is not WaitingForLength.
    fn pop_length(&mut self) -> Result<Option<u64>, Error> {
        const UNEXPECTED_EOF: io::ErrorKind = io::ErrorKind::UnexpectedEof;

        debug_assert_eq!(FramerState::WaitingForLength, self.state);

        let mut available = &self.buffer[self.consumed..];

        match leb128::read::unsigned(&mut available) {
            Ok(length) => {
                let consumed = self.buffer.len() - self.consumed - available.len();
                self.consumed += consumed;

                Ok(Some(length))
            },
            Err(leb128::read::Error::IoError(e)) if e.kind() == UNEXPECTED_EOF => Ok(None),
            Err(e) => Err(e.into()),
        }
    }

    //  Pops the actual frame, according to the length.
    //
    //  # Panics
    //
    //  In Debug, if called when state is not WaitingForMessage(length).
    fn pop_message(&mut self, length: NonZeroU64) -> Result<Option<&[u8]>, Error> {
        debug_assert_eq!(FramerState::WaitingForMessage(length), self.state);

        let length = length.get().try_into().map_err(|_| Error::Overflow)?;

        let Some((frame, _)) = self.buffer[self.consumed..].split_at_checked(length) else {
            return Ok(None);
        };

        self.state = FramerState::WaitingForLength;
        self.consumed += frame.len();

        Ok(Some(frame))
    }
}

#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
enum FramerState {
    #[default]
    WaitingForLength,
    WaitingForBytes(NonZeroU64),
}