liminal-cli 0.3.1

//! A raw test publisher uses the real server's canonical frames and refusals.

use std::error::Error;
use std::io::{Read, Write};
use std::net::TcpStream;

use liminal::protocol::{
    CausalContext, Frame, MessageEnvelope, ProtocolError, ProtocolVersion, SchemaId, decode,
    encode, encoded_len,
};

pub(super) struct Publisher {
    socket: TcpStream,
    buffer: Vec<u8>,
}

impl Publisher {
    pub fn connect(address: &str, credential: &[u8]) -> Result<Self, Box<dyn Error>> {
        let mut publisher = Self {
            socket: TcpStream::connect(address)?,
            buffer: Vec::new(),
        };
        publisher.write(&Frame::Connect {
            flags: 0,
            min_version: ProtocolVersion::new(1, 0),
            max_version: ProtocolVersion::new(1, 2),
            auth_token: credential.to_vec(),
        })?;
        match publisher.read()? {
            Frame::ConnectAck { .. } => Ok(publisher),
            _ => Err("the fixture publisher was not admitted".into()),
        }
    }

    pub fn publish(&mut self, subject: &str, payload: Vec<u8>) -> Result<Frame, Box<dyn Error>> {
        let envelope = MessageEnvelope::new(
            SchemaId::new([0; SchemaId::WIRE_LEN]),
            CausalContext::independent(),
            payload,
        );
        self.write(&Frame::new_publish(1, subject, envelope)?)?;
        self.read()
    }

    pub fn subscribe(&mut self, subject: &str) -> Result<Frame, Box<dyn Error>> {
        self.write(&Frame::Subscribe {
            flags: 0,
            stream_id: 1,
            channel: subject.to_owned(),
            accepted_schemas: Vec::new(),
            max_in_flight: 1,
        })?;
        self.read()
    }

    fn write(&mut self, frame: &Frame) -> Result<(), Box<dyn Error>> {
        let mut bytes = vec![0; encoded_len(frame)?];
        let count = encode(frame, &mut bytes)?;
        self.socket.write_all(
            bytes
                .get(..count)
                .ok_or("the encoded frame size is invalid")?,
        )?;
        Ok(self.socket.flush()?)
    }

    fn read(&mut self) -> Result<Frame, Box<dyn Error>> {
        loop {
            match decode(&self.buffer) {
                Ok((frame, count)) => {
                    self.buffer.drain(..count);
                    return Ok(frame);
                }
                Err(
                    ProtocolError::IncompleteHeader { .. } | ProtocolError::TruncatedPayload { .. },
                ) => {}
                Err(_) => return Err("the server sent an invalid canonical frame".into()),
            }
            // This is a transport read chunk, not a delivery or test deadline.
            let mut chunk = [0; 8192];
            let count = self.socket.read(&mut chunk)?;
            if count == 0 {
                return Err("the server closed before returning a canonical frame".into());
            }
            self.buffer
                .extend_from_slice(chunk.get(..count).ok_or("the read count is invalid")?);
        }
    }
}