liminal-cli 0.3.1

//! Receiver ownership follows explicit unsubscribe, EOF and stream-end signals.

use super::channel::{self, Batch};
use super::channel_join::Join;
use super::listen::{Listens, Pending, Subject};
use super::listen_config::ListenConfig;
use super::writer::Writer;
use std::io::{self, Write};
use std::sync::Arc;
use std::sync::atomic::Ordering;
use std::thread::{Scope, ScopedJoinHandle};

pub(super) type Reader<'scope> = (Arc<Subject>, ScopedJoinHandle<'scope, io::Result<()>>);

#[derive(Clone)]
pub(super) struct Context {
    pub config: ListenConfig,
    pub device: String,
    pub record_bytes: usize,
}

pub(super) fn spawn<'scope, 'env, O: Write + Send + 'scope>(
    scope: &'scope Scope<'scope, 'env>,
    readers: &mut Vec<Reader<'scope>>,
    listens: &Arc<Listens>,
    writer: &Arc<Writer<O>>,
    context: &Context,
) -> io::Result<()> {
    for pending in listens.pending().map_err(io::Error::other)? {
        let Pending {
            subject,
            tap,
            delivery,
            join,
            tap_close,
        } = pending;
        let joined = Arc::clone(&join);
        let watched = Arc::clone(&subject);
        let config = context.config.clone();
        let payload = std::thread::Builder::new()
            .name("liminal-mcp-channel-join".to_owned())
            .spawn_scoped(scope, move || {
                watch(&delivery, &joined, &watched, &config, &tap_close)
            })?;
        listens.receivers.fetch_add(1, Ordering::SeqCst);
        readers.push((Arc::clone(&subject), payload));
        let tracked = Arc::clone(&subject);
        let owned_listens = Arc::clone(listens);
        let writer = Arc::clone(writer);
        let context = context.clone();
        let input = Input { subject, tap, join };
        let reader = std::thread::Builder::new()
            .name("liminal-mcp-channel".to_owned())
            .spawn_scoped(scope, move || {
                receive(&input, &owned_listens, &writer, &context)
            })?;
        listens.receivers.fetch_add(1, Ordering::SeqCst);
        readers.push((tracked, reader));
    }
    Ok(())
}

struct Input {
    subject: Arc<Subject>,
    tap: crate::rail_link::Subscription,
    join: Arc<Join>,
}

fn watch(
    stream: &crate::rail_link::Subscription,
    join: &Join,
    subject: &Subject,
    config: &ListenConfig,
    tap_close: &super::listen::Close,
) -> io::Result<()> {
    let result: Result<(), String> = (|| {
        while subject.active.load(Ordering::SeqCst) {
            let delivery = stream.recv().map_err(super::channel_source::words)?;
            join.offer(&delivery, config)?;
        }
        Ok(())
    })();
    if let Err(reason) = result {
        let missing_id = reason == "the channel delivery has no id to join to the tap";
        let ended = join.end(reason);
        // An inline tap payload needs no delivery join. Its record decides
        // whether an absent delivery id is terminal; it is never guessed.
        let closed = if missing_id {
            Ok(())
        } else {
            tap_close().map_err(super::channel_source::words)
        };
        match (ended, closed) {
            (Ok(()), Ok(())) => {}
            (Err(first), Err(second)) => {
                return Err(io::Error::other(format!("{first}; {second}")));
            }
            (Err(reason), Ok(())) | (Ok(()), Err(reason)) => return Err(io::Error::other(reason)),
        }
    }
    Ok(())
}

fn receive<O: Write>(
    pending: &Input,
    listens: &Listens,
    writer: &Writer<O>,
    context: &Context,
) -> io::Result<()> {
    let result = records(pending, writer, context);
    if let Err(reason) = result
        && pending.subject.active.swap(false, Ordering::SeqCst)
    {
        let direct = reason.to_string();
        let reason = match pending.join.reason().map_err(io::Error::other)? {
            Some(reason)
                if direct == "subscription receive failed: the subscription was closed" =>
            {
                reason
            }
            Some(reason) if reason != direct => format!("{direct}; payload stream ended: {reason}"),
            Some(_) | None => direct,
        };
        let ended = listens.ended(&pending.subject);
        let reason = match ended {
            Ok(()) => reason,
            Err(words) => format!("{reason}; {words}"),
        };
        writer.write(
            channel::outcome(
                &pending.subject.name,
                "ended",
                &format!("listening on {} ended: {reason}", pending.subject.name),
            ),
            None,
            true,
        )?;
    }
    Ok(())
}

fn records<O: Write>(pending: &Input, writer: &Writer<O>, context: &Context) -> io::Result<()> {
    while pending.subject.active.load(Ordering::SeqCst) {
        let delivery = pending
            .tap
            .recv()
            .map_err(super::channel_source::words)
            .map_err(io::Error::other)?;
        let batch = Batch::read(
            delivery.payload(),
            &pending.subject.name,
            &context.config,
            context.record_bytes,
        )
        .map_err(io::Error::other)?;
        for message in batch.messages.iter().filter(|message| message.delivered) {
            let words = pending
                .join
                .content(message, &context.config)
                .map_err(io::Error::other)?;
            writer.channel(
                message.notification(&context.device, &words),
                &pending.subject,
            )?;
        }
        if batch.dropped > 0 {
            writer.channel(
                channel::outcome(
                    &pending.subject.name,
                    "dropped",
                    &format!(
                        "{} messages on {} were not kept",
                        batch.dropped, pending.subject.name
                    ),
                ),
                &pending.subject,
            )?;
        }
    }
    Ok(())
}

pub(super) fn join_closed(readers: &mut Vec<Reader<'_>>, listens: &Listens) -> io::Result<()> {
    let mut n = 0;
    while n < readers.len() {
        if readers[n].0.active.load(Ordering::SeqCst) {
            n += 1;
        } else {
            joined(readers.swap_remove(n).1, listens)?;
        }
    }
    Ok(())
}

pub(super) fn joined(
    reader: ScopedJoinHandle<'_, io::Result<()>>,
    listens: &Listens,
) -> io::Result<()> {
    let result = reader
        .join()
        .map_err(|_| io::Error::other("a channel receiver panicked"));
    listens.receivers.fetch_sub(1, Ordering::SeqCst);
    result?
}