liminal-sdk 0.19.1

//! The Rust SDK's reply helper (Fable 4c), pinned against a scripted
//! wire-level server that records the causal parent of every publish.
//!
//! A reply through [`RemoteChannelHandle::reply_to`] names the request it
//! answers as `<boot>:<id>`, the pair a 1.2 server's Deliver carried, so the
//! server's tap can pair the two and measure the dwell between them. A request
//! that carried no id gives a reply with no parent, never an invented one.
#![cfg(feature = "std")]

use std::error::Error;
use std::net::{TcpListener, TcpStream};
use std::sync::mpsc::{Receiver, Sender, channel};
use std::thread::JoinHandle;

use liminal::protocol::{
    DELIVER_MESSAGE_ID_VERSION, DeliveredId, Frame, MessageId, ProtocolError, decode, encode,
    encoded_len,
};
use liminal_sdk::{
    ConnectionPoolConfig, RemoteChannelHandle, RemoteConfig, SchemaMetadata, SchemaValidate,
};
use serde::Serialize;

/// The channel the replies are published on.
const CHANNEL: &str = "orders.reply";

/// The request the replies answer, as its Deliver carried it.
const REQUEST: DeliveredId = DeliveredId {
    boot: 1_791_328_512_345,
    id: 42,
};

#[derive(Serialize)]
struct Answer {
    id: u64,
}

impl SchemaValidate for Answer {
    fn schema_metadata() -> SchemaMetadata {
        SchemaMetadata::new("test.answer", "1", br#"{"type":"object"}"#.as_slice())
    }
}

/// A single-connection scripted server that records each publish's causal
/// parent and acks it.
struct ScriptedServer {
    addr: String,
    parents: Receiver<Option<MessageId>>,
    handle: JoinHandle<()>,
}

impl ScriptedServer {
    fn spawn() -> Result<Self, Box<dyn Error>> {
        let listener = TcpListener::bind("127.0.0.1:0")?;
        let addr = listener.local_addr()?.to_string();
        let (parents_tx, parents) = channel();
        let handle = std::thread::spawn(move || {
            if let Ok((stream, _)) = listener.accept() {
                serve(stream, &parents_tx);
            }
        });
        Ok(Self {
            addr,
            parents,
            handle,
        })
    }

    /// Waits for the server to see its connection close, failing if it
    /// panicked.
    fn finish(self) -> Result<(), Box<dyn Error>> {
        self.handle
            .join()
            .map_err(|_| "the scripted server panicked".into())
    }
}

fn serve(mut stream: TcpStream, parents: &Sender<Option<MessageId>>) {
    let mut buffer: Vec<u8> = Vec::new();
    loop {
        match decode(&buffer) {
            Ok((frame, consumed)) => {
                buffer.drain(..consumed);
                let reply = match frame {
                    Frame::Connect { .. } => Frame::ConnectAck {
                        flags: 0,
                        selected_version: DELIVER_MESSAGE_ID_VERSION,
                        capabilities: 0,
                    },
                    Frame::Publish {
                        stream_id,
                        envelope,
                        ..
                    } => {
                        if parents.send(envelope.causal_context.parent_id).is_err() {
                            return;
                        }
                        Frame::PublishAck {
                            flags: 0,
                            stream_id,
                            message_id: 1,
                            durable_position: None,
                        }
                    }
                    _ => return,
                };
                if write_frame(&mut stream, &reply).is_err() {
                    return;
                }
            }
            Err(
                ProtocolError::IncompleteHeader { .. } | ProtocolError::TruncatedPayload { .. },
            ) => {
                let mut chunk = [0_u8; 4096];
                match std::io::Read::read(&mut stream, &mut chunk) {
                    Ok(0) | Err(_) => return,
                    Ok(read) => match chunk.get(..read) {
                        Some(received) => buffer.extend_from_slice(received),
                        None => return,
                    },
                }
            }
            Err(_) => return,
        }
    }
}

fn write_frame(stream: &mut TcpStream, frame: &Frame) -> Result<(), Box<dyn Error>> {
    let mut bytes = vec![0_u8; encoded_len(frame)?];
    let written = encode(frame, &mut bytes)?;
    let encoded = bytes.get(..written).ok_or("encoder byte count invalid")?;
    std::io::Write::write_all(stream, encoded)?;
    Ok(())
}

fn connect(server: &ScriptedServer) -> Result<RemoteChannelHandle, Box<dyn Error>> {
    let config = RemoteConfig::new(
        server.addr.clone(),
        CHANNEL,
        "conversation".to_owned(),
        ConnectionPoolConfig::new(1, 10, 16),
    )?
    .connect_tcp()?;
    Ok(RemoteChannelHandle::new(&config)?)
}

/// 4c: a reply through the helper names its request as `<boot>:<id>`.
#[test]
fn a_reply_names_its_request_by_boot_and_id() -> Result<(), Box<dyn Error>> {
    let server = ScriptedServer::spawn()?;
    let handle = connect(&server)?;
    handle.reply_to(Some(REQUEST), &Answer { id: 1 })?;
    assert_eq!(
        server.parents.recv()?,
        Some(MessageId::from("1791328512345:42"))
    );
    drop(handle);
    server.finish()
}

/// 4c: a request that carried no id (an old server, or a durable message
/// replayed from the store) gives a reply with no parent.
#[test]
fn a_reply_to_a_request_without_an_id_names_no_parent() -> Result<(), Box<dyn Error>> {
    let server = ScriptedServer::spawn()?;
    let handle = connect(&server)?;
    handle.reply_to(None, &Answer { id: 1 })?;
    assert_eq!(server.parents.recv()?, None);
    drop(handle);
    server.finish()
}