Quelle  codec.rs

// Copyright © 2017 Mozilla Foundation
//
// This program is made available under an ISC-style license.  See the
// accompanying file LICENSE for details

//! `Encoder`s and `Decoder`s from items to/from `BytesMut` buffers.

// The assert in LengthDelimitedCodec::decode triggers this clippy warning but
// requires upgrading the workspace to Rust 2021 to resolve.
// This should be fixed in Rust 1.68, after which the following `allow` can be deleted.
#![allow(clippy::uninlined_format_args)]

use bincode::Options;
use byteorder::{ByteOrder, LittleEndian};
use bytes::{Buf, BufMut, BytesMut};
use serde::de::DeserializeOwned;
use serde::ser::Serialize;
use std::convert::TryInto;
use std::fmt::Debug;
use std::io;
use std::marker::PhantomData;
use std::mem::size_of;

////////////////////////////////////////////////////////////////////////////////
// Split buffer into size delimited frames - This appears more complicated than
// might be necessary due to handling the possibility of messages being split
// across reads.

pub trait Codec {
    /// The type of items to be encoded into byte buffer
    type In;

    /// The type of items to be returned by decoding from byte buffer
    type Out;

    /// Attempts to decode a frame from the provided buffer of bytes.
    fn decode(&mut self, buf: &mut BytesMut) -> io::Result<Option<Self::Out>>;

    /// A default method available to be called when there are no more bytes
    /// available to be read from the I/O.
    fn decode_eof(&mut self, buf: &mut BytesMut) -> io::Result<Self::Out> {
        match self.decode(buf)? {
            Some(frame) => Ok(frame),
            None => Err(io::Error::new(
                io::ErrorKind::Other,
                "bytes remaining on stream",
            )),
        }
    }

    /// Encodes a frame into the buffer provided.
    fn encode(&mut self, msg: Self::In, buf: &mut BytesMut) -> io::Result<()>;
}

/// Codec based upon bincode serialization
///
/// Messages that have been serialized using bincode are prefixed with
/// the length of the message to aid in deserialization, so that it's
/// known if enough data has been received to decode a complete
/// message.
pub struct LengthDelimitedCodec<In, Out> {
    state: State,
    encode_buf: Vec<u8>,
    __in: PhantomData<In>,
    __out: PhantomData<Out>,
}

enum State {
    Length,
    Data(u32),
}

const MAX_MESSAGE_LEN: u32 = 1024 * 1024;
const MAGIC: u64 = 0xa4d1_019c_c910_1d4a;
const HEADER_LEN: usize = size_of::<u32>() + size_of::<u64>();

impl<In, Out> Default for LengthDelimitedCodec<In, Out> {
    fn default() -> Self {
        Self {
            state: State::Length,
            encode_buf: Vec::with_capacity(crate::ipccore::IPC_CLIENT_BUFFER_SIZE),
            __in: PhantomData,
            __out: PhantomData,
        }
    }
}

impl<In, Out> LengthDelimitedCodec<In, Out> {
    // Lengths are encoded as little endian u32
    fn decode_length(buf: &mut BytesMut) -> Option<u32> {
        if buf.len() < HEADER_LEN {
            // Not enough data
            return None;
        }

        let magic = LittleEndian::read_u64(&buf[0..8]);
        assert_eq!(magic, MAGIC);

        // Consume the length field
        let n = LittleEndian::read_u32(&buf[8..12]);
        buf.advance(HEADER_LEN);
        Some(n)
    }

    fn decode_data(buf: &mut BytesMut, n: u32) -> io::Result<Option<Out>>
    where
        Out: DeserializeOwned + Debug,
    {
        let n = n.try_into().unwrap();

        // At this point, the buffer has already had the required capacity
        // reserved. All there is to do is read.
        if buf.len() < n {
            return Ok(None);
        }

        trace!("Attempting to decode");
        let msg = bincode::options()
            .with_limit(MAX_MESSAGE_LEN as u64)
            .deserialize::<Out>(&buf[..n])
            .map_err(|e| match *e {
                bincode::ErrorKind::Io(e) => e,
                _ => io::Error::new(io::ErrorKind::Other, *e),
            })?;
        buf.advance(n);

        trace!("... Decoded {:?}", msg);
        Ok(Some(msg))
    }
}

impl<In, Out> Codec for LengthDelimitedCodec<In, Out>
where
    In: Serialize + Debug,
    Out: DeserializeOwned + Debug,
{
    type In = In;
    type Out = Out;

    fn decode(&mut self, buf: &mut BytesMut) -> io::Result<Option<Self::Out>> {
        let n = match self.state {
            State::Length => {
                match Self::decode_length(buf) {
                    Some(n) => {
                        assert!(
                            n <= MAX_MESSAGE_LEN,
                            "assertion failed: {} <= {}",
                            n,
                            MAX_MESSAGE_LEN
                        );
                        self.state = State::Data(n);

                        // Ensure that the buffer has enough space to read the
                        // incoming payload
                        buf.reserve(n.try_into().unwrap());

                        n
                    }
                    None => return Ok(None),
                }
            }
            State::Data(n) => n,
        };

        match Self::decode_data(buf, n)? {
            Some(data) => {
                // Update the decode state
                self.state = State::Length;

                // Make sure the buffer has enough space to read the next length header.
                buf.reserve(HEADER_LEN);

                Ok(Some(data))
            }
            None => Ok(None),
        }
    }

    fn encode(&mut self, item: Self::In, buf: &>mut BytesMut) -> io::Result<()> {
        trace!("Attempting to encode");

        self.encode_buf.clear();
        if let Err(e) = bincode::options()
            .with_limit(MAX_MESSAGE_LEN as u64)
            .serialize_into::<_, Self::In>(&mut self.encode_buf, &item)
        {
            trace!("message encode failed: {:?}", *e);
            match *e {
                bincode::ErrorKind::Io(e) => return Err(e),
                _ => return Err(io::Error::new(io::ErrorKind::Other, *e)),
            }
        }

        let encoded_len = self.encode_buf.len();
        assert!(encoded_len <= MAX_MESSAGE_LEN as usize);
        buf.reserve(encoded_len + HEADER_LEN);
        buf.put_u64_le(MAGIC);
        buf.put_u32_le(encoded_len.try_into().unwrap());
        buf.extend_from_slice(&self.encode_buf);

        Ok(())
    }
}