haematite-cli 0.21.7

use std::io;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::mpsc::{self, Receiver, Sender};
use std::thread::{self, JoinHandle};

use serde_json::Value;

use super::remote::{Bridge, MESSAGE_BYTES, Output};
use super::remote_http::REQUEST_FAILED;
use super::stdio::read_line_bounded;

type Reader = (Receiver<io::Result<Vec<u8>>>, JoinHandle<io::Result<()>>);

pub(super) fn spawn(bridge: Bridge, output: Output, active: Arc<AtomicBool>) -> io::Result<Reader> {
    let (sender, receiver) = mpsc::channel();
    let reader = thread::Builder::new()
        .name("haematite-mcp-stdin".into())
        .spawn(move || {
            if let Err(error) = read(&sender, &bridge, &output, &active) {
                sender.send(Err(error)).map_err(io::Error::other)?;
            }
            Ok(())
        })?;
    Ok((receiver, reader))
}

fn read(
    sender: &Sender<io::Result<Vec<u8>>>,
    bridge: &Bridge,
    output: &Output,
    active: &AtomicBool,
) -> io::Result<()> {
    let mut input = io::stdin().lock();
    let mut notifications: Vec<JoinHandle<io::Result<()>>> = Vec::new();
    while let Some(line) = read_line_bounded(&mut input, MESSAGE_BYTES)? {
        let notification = line.len() <= MESSAGE_BYTES
            && serde_json::from_slice::<Value>(&line).is_ok_and(|value| {
                value.get("method").is_some_and(Value::is_string) && value.get("id").is_none()
            });
        if active.load(Ordering::Acquire) && notification {
            // A long session keeps no handle of a notification already done.
            let mut running = Vec::with_capacity(notifications.len() + 1);
            for worker in notifications {
                if worker.is_finished() {
                    joined(worker)?;
                } else {
                    running.push(worker);
                }
            }
            notifications = running;
            let bridge = bridge.clone();
            let state = bridge.snapshot()?;
            let output = output.clone();
            let worker = thread::Builder::new()
                .name("haematite-mcp-notification".into())
                .spawn(move || bridge.handle(&line, &output, &state))
                .inspect_err(|_| {
                    eprintln!("haematite_mcp_remote_notification_refused: {REQUEST_FAILED}");
                })?;
            notifications.push(worker);
        } else {
            sender.send(Ok(line)).map_err(io::Error::other)?;
        }
    }
    // EOF waits for each accepted notification to finish and report its outcome.
    for notification in notifications {
        joined(notification)?;
    }
    Ok(())
}

fn joined(worker: JoinHandle<io::Result<()>>) -> io::Result<()> {
    worker
        .join()
        .map_err(|_| io::Error::other("MCP notification thread panicked"))?
}