//! A client for `broker::handle` and `admin::handle` over a socket pair, and a `Broker` built on
//! a `Rig`. Do not edit.
//!
//! Included with `#[path = "support/client.rs"] mod client;`, beside `rig`, `runtime`, `sink`
//! and `tmp`.

#![allow(dead_code)] // each test file uses a different part of this module

use std::os::unix::net::UnixStream;
use std::sync::Arc;
use std::time::Duration;

use brokerd::broker::{self, Broker};
use proto::{Envelope, Message, PROTOCOL_VERSION, ToolRequest};

use crate::rig::Rig;
use crate::runtime::{Recording, Shared};

pub trait Serve {
    /// A `Broker` on this rig's home, with `runtime` and the rig's flaky sink and log.
    fn broker(&self, runtime: &Arc<Recording>) -> Arc<Broker>;
}

impl Serve for Rig {
    fn broker(&self, runtime: &Arc<Recording>) -> Arc<Broker> {
        let runtime = Box::new(Shared(Arc::clone(runtime)));
        Arc::new(Broker::new(
            self.cfg.clone(),
            self.ledger(),
            runtime,
            self.lines.sink(),
        ))
    }
}

/// A connection to `handler` running on its own thread, with `msg` already sent under `id`.
pub fn open(
    broker: &Arc<Broker>,
    handler: fn(UnixStream, &Broker),
    id: u64,
    msg: Message,
) -> UnixStream {
    let (mut client, server) = UnixStream::pair().unwrap();
    let broker = Arc::clone(broker);
    std::thread::spawn(move || handler(server, &broker));
    let env = Envelope {
        v: PROTOCOL_VERSION,
        id,
        r#final: true,
        msg,
    };
    proto::write_frame(&mut client, &env).unwrap();
    client
}

/// The next frame, waiting at most ten seconds. Once the handler has closed its end, macOS
/// refuses the timeout with EINVAL (22); the frame is buffered by then and the read cannot block.
pub fn next(stream: &mut UnixStream) -> Envelope {
    if let Err(e) = stream.set_read_timeout(Some(Duration::from_secs(10))) {
        let peer_closed = cfg!(target_vendor = "apple") && e.raw_os_error() == Some(22);
        assert!(peer_closed, "set_read_timeout: {e}");
    }
    proto::read_frame(stream).unwrap()
}

/// Sends a tool request to `broker::handle` and reads every frame up to the final one.
pub fn call(broker: &Arc<Broker>, req: ToolRequest) -> Vec<Envelope> {
    let id = req.call.0;
    let mut stream = open(broker, broker::handle, id, Message::ToolRequest(req));
    let mut frames = Vec::new();
    loop {
        let env = next(&mut stream);
        let last = env.r#final;
        frames.push(env);
        if last {
            return frames;
        }
    }
}