oc2_hlapi_v2 0.2.0

//! HLAPI codecs
//!
//! For now, only one codec is provided, the codec that all base OC2 code uses for HLAPI messages.

use std::error::Error;
use std::fmt::Debug;
use std::io::{self, BufRead, BufReader, Read};
use std::mem;
use std::time::{Duration, Instant};

use crate::messages::{BlobDescriptor, Event, Request, Response};
use crate::{Codec, Stream, Transport};

use serde_json::Value as JsonValue;
use thiserror::Error;

/// The base three HLAPI streams
#[derive(Debug)]
pub struct BaseStreams<T> {
    /// The RPC stream, for requests and responses
    pub rpc: T,
    /// The blob stream
    pub blob: T,
    /// The event stream
    pub event: T,
}

/// The base HLAPI codec
///
/// Three streams, one for requests and responses, one for events, and one
/// for binary blobs. Messages are (de)serialized to/from JSON, and framed with a null byte on either
/// side.
#[derive(Debug)]
pub struct BaseCodec<S, T: Transport<Streams = BaseStreams<S>>> {
    transport: T,
    rpc: BufReader<S>,
    blob: S,
    event: BufReader<S>,
    partial_resp: Vec<u8>,
    partial_event: Vec<u8>,
}

impl<S: Stream, T: Transport<Streams = BaseStreams<S>>> BaseCodec<S, T> {
    /// Create a new codec using the given transport
    pub fn new(mut transport: T) -> Result<Self, T::Error> {
        let streams = transport.open_streams()?;
        Ok(Self {
            transport,
            rpc: BufReader::new(streams.rpc),
            blob: streams.blob,
            event: BufReader::new(streams.event),
            partial_resp: Vec::new(),
            partial_event: Vec::new(),
        })
    }

    fn reopen_streams(&mut self) -> Result<(), BaseCodecError<T::Error>> {
        let streams = self
            .transport
            .open_streams()
            .map_err(BaseCodecError::Transport)?;
        self.rpc = BufReader::new(streams.rpc);
        self.blob = streams.blob;
        self.event = BufReader::new(streams.event);
        self.partial_resp.clear();
        self.partial_event.clear();
        Ok(())
    }
}

/// Errors produced by the standard codec
#[derive(Debug, Error)]
#[non_exhaustive]
pub enum BaseCodecError<E: Error> {
    /// An I/O error
    #[error("I/O error: {0}")]
    Io(#[from] io::Error),
    /// A serde error
    #[error("Serde error: {0}")]
    Serde(#[from] serde_json::Error),
    /// An error produced by the transport
    #[error("Transport error: {0}")]
    Transport(#[source] E),
    /// Too big of a request was given to `send_request`
    #[error("Request too big")]
    RequestTooBig,
    /// `read_blob` encountered a checksum error
    #[error("Blob checksum invalid")]
    BlobChecksumInvalid,
}

impl<S: Stream, T: Transport<Streams = BaseStreams<S>>> BaseCodec<S, T> {
    fn send_request_raw(
        &mut self,
        request: &Request,
        blobs: &[&[u8]],
    ) -> Result<(), <Self as Codec>::Error> {
        let mut buf = Vec::new();
        buf.push(0);
        serde_json::to_writer(&mut buf, &request)?;
        buf.push(0);
        // Whether the null bytes count as part of the frame length is unclear, so assume they do.
        if buf.len() > 4096 {
            return Err(BaseCodecError::RequestTooBig);
        }
        // Only send the blobs once the request validates, otherwise they could cause a desync if
        // the blobs are sent without the request being sent.
        for blob in blobs {
            self.blob.write_all(blob)?;
        }
        self.rpc.get_mut().write_all(&buf)?;
        Ok(())
    }

    fn read_response_raw(
        &mut self,
        timeout: Option<Duration>,
    ) -> Result<Option<Response>, <Self as Codec>::Error> {
        let deadline = timeout.map(|x| Instant::now() + x);
        loop {
            let remaining = deadline.map(|d| d.saturating_duration_since(Instant::now()));
            if remaining == Some(Duration::ZERO) {
                self.rpc.get_ref().set_nonblocking(true)?;
                self.rpc.get_ref().set_read_timeout(None)?;
            } else {
                self.rpc.get_ref().set_nonblocking(false)?;
                self.rpc.get_ref().set_read_timeout(remaining)?;
            }
            match self.rpc.read_until(0, &mut self.partial_resp) {
                Ok(0) => {
                    self.reopen_streams()?;
                    return Err(io::Error::new(
                        io::ErrorKind::UnexpectedEof,
                        "EOF when reading response",
                    )
                    .into());
                }
                Ok(_) => (),
                Err(e) => {
                    if e.kind() == io::ErrorKind::WouldBlock || e.kind() == io::ErrorKind::TimedOut
                    {
                        return Ok(None);
                    } else {
                        return Err(e.into());
                    }
                }
            }
            // If we do not have a full frame, continue
            if self.partial_resp.last() != Some(&0) {
                continue;
            }
            self.partial_resp.pop();
            let frame = mem::take(&mut self.partial_resp);
            if frame.is_empty() {
                // Ignore empty frames
                continue;
            }
            // Ignore non-JSON frames
            if let Ok(value) = serde_json::from_slice::<JsonValue>(&frame) {
                return Ok(Some(serde_json::from_value(value)?));
            }
        }
    }

    fn read_event_raw(
        &mut self,
        timeout: Option<Duration>,
    ) -> Result<Option<Event>, <Self as Codec>::Error> {
        let deadline = timeout.map(|x| Instant::now() + x);
        loop {
            let remaining = deadline.map(|d| d.saturating_duration_since(Instant::now()));
            if remaining == Some(Duration::ZERO) {
                self.event.get_ref().set_nonblocking(true)?;
                self.event.get_ref().set_read_timeout(None)?;
            } else {
                self.event.get_ref().set_nonblocking(false)?;
                self.event.get_ref().set_read_timeout(remaining)?;
            }
            match self.event.read_until(0, &mut self.partial_event) {
                Ok(0) => {
                    self.reopen_streams()?;
                    return Err(io::Error::new(
                        io::ErrorKind::UnexpectedEof,
                        "EOF when reading event",
                    )
                    .into());
                }
                Ok(_) => (),
                Err(e) => {
                    if e.kind() == io::ErrorKind::WouldBlock || e.kind() == io::ErrorKind::TimedOut
                    {
                        return Ok(None);
                    } else {
                        return Err(e.into());
                    }
                }
            }
            // If we do not have a full frame, continue
            if self.partial_event.last() != Some(&0) {
                continue;
            }
            self.partial_event.pop();
            let frame = mem::take(&mut self.partial_event);
            if frame.is_empty() {
                // Ignore empty frames
                continue;
            }
            // Ignore unparseable frames
            if let Ok(event) = serde_json::from_slice::<Event>(&frame) {
                return Ok(Some(event));
            }
        }
    }

    fn read_blob_raw(
        &mut self,
        descriptor: BlobDescriptor,
    ) -> Result<Vec<u8>, <Self as Codec>::Error> {
        let mut buf = vec![0; descriptor.length as usize];
        self.blob.read_exact(&mut buf)?;
        if !descriptor.verify_checksum(&buf) {
            return Err(BaseCodecError::BlobChecksumInvalid);
        }
        Ok(buf)
    }
}

impl<S: Stream, T: Transport<Streams = BaseStreams<S>>> Codec for BaseCodec<S, T> {
    type Error = BaseCodecError<T::Error>;

    fn send_request(&mut self, request: Request, blobs: &[&[u8]]) -> Result<(), Self::Error> {
        match self.send_request_raw(&request, blobs) {
            Ok(()) => Ok(()),
            Err(BaseCodecError::Io(e)) => {
                if e.kind() == io::ErrorKind::BrokenPipe {
                    self.reopen_streams()?;
                }
                Err(BaseCodecError::Io(e))
            }
            Err(e) => Err(e),
        }
    }

    fn read_response(
        &mut self,
        timeout: Option<Duration>,
    ) -> Result<Option<Response>, Self::Error> {
        match self.read_response_raw(timeout) {
            Ok(x) => Ok(x),
            Err(BaseCodecError::Io(e)) => {
                if e.kind() == io::ErrorKind::BrokenPipe {
                    self.reopen_streams()?;
                }
                Err(BaseCodecError::Io(e))
            }
            Err(e) => Err(e),
        }
    }

    fn read_event(&mut self, timeout: Option<Duration>) -> Result<Option<Event>, Self::Error> {
        match self.read_event_raw(timeout) {
            Ok(x) => Ok(x),
            Err(BaseCodecError::Io(e)) => {
                if e.kind() == io::ErrorKind::BrokenPipe {
                    self.reopen_streams()?;
                    self.read_event(timeout)
                } else {
                    Err(BaseCodecError::Io(e))
                }
            }
            Err(e) => Err(e),
        }
    }

    fn read_blob(&mut self, descriptor: BlobDescriptor) -> Result<Vec<u8>, Self::Error> {
        match self.read_blob_raw(descriptor) {
            Ok(x) => Ok(x),
            Err(BaseCodecError::Io(e)) => {
                if e.kind() == io::ErrorKind::BrokenPipe {
                    self.reopen_streams()?;
                }
                Err(BaseCodecError::Io(e))
            }
            Err(e) => Err(e),
        }
    }

    fn clear_response_stream(&mut self) -> Result<(), Self::Error> {
        self.partial_resp.clear();
        self.rpc.get_ref().set_nonblocking(true)?;
        let mut buf = [0; 4096];
        loop {
            match self.rpc.read(&mut buf) {
                Ok(amount_read) => {
                    if amount_read == 0 {
                        self.reopen_streams()?;
                        return Err(io::Error::new(
                            io::ErrorKind::UnexpectedEof,
                            "EOF when clearing response stream",
                        )
                        .into());
                    }
                }
                Err(e) => {
                    if e.kind() == io::ErrorKind::WouldBlock {
                        break;
                    } else if e.kind() == io::ErrorKind::BrokenPipe {
                        self.reopen_streams()?;
                    }
                    return Err(e.into());
                }
            }
        }
        self.rpc.get_ref().set_nonblocking(false)?;
        Ok(())
    }

    fn clear_blob_stream(&mut self) -> Result<(), Self::Error> {
        self.blob.set_nonblocking(true)?;
        let mut buf = [0; 4096];
        loop {
            match self.blob.read(&mut buf) {
                Ok(amount_read) => {
                    if amount_read == 0 {
                        self.reopen_streams()?;
                        return Err(io::Error::new(
                            io::ErrorKind::UnexpectedEof,
                            "EOF when clearing blob stream",
                        )
                        .into());
                    }
                }
                Err(e) => {
                    if e.kind() == io::ErrorKind::WouldBlock {
                        break;
                    } else if e.kind() == io::ErrorKind::BrokenPipe {
                        self.reopen_streams()?;
                    }
                    return Err(e.into());
                }
            }
        }
        self.blob.set_nonblocking(false)?;
        Ok(())
    }
}