liminal-cli 0.3.3

//! Signal-driven tap fixtures: one held record, consumed before the feeder returns.

use std::collections::BTreeMap;
use std::sync::{Arc, Condvar, Mutex};

use serde_json::Value;

use super::RailError;

type Record = Result<Value, RailError>;

#[derive(Debug, Default)]
struct SlotState {
    record: Option<Record>,
    closed: bool,
}

#[derive(Debug, Default)]
pub struct Slot {
    state: Mutex<SlotState>,
    changed: Condvar,
}

impl Slot {
    pub fn receive(&self) -> Result<Value, RailError> {
        let mut state = self.state.lock().map_err(poisoned)?;
        loop {
            if state.closed {
                return Err(RailError::Refused("fixture closed".to_owned()));
            }
            if let Some(record) = state.record.take() {
                self.changed.notify_all();
                return record;
            }
            state = self.changed.wait(state).map_err(poisoned)?;
        }
    }

    pub fn close(&self) -> Result<(), RailError> {
        self.state.lock().map_err(poisoned)?.closed = true;
        self.changed.notify_all();
        Ok(())
    }
}

#[derive(Debug, Default)]
struct State {
    slots: BTreeMap<String, Arc<Slot>>,
    refused: BTreeMap<String, String>,
    opened: Vec<String>,
    history: Vec<Arc<Slot>>,
}

#[derive(Clone, Debug, Default)]
pub struct Harness(Arc<Mutex<State>>);

fn poisoned<T>(_: T) -> RailError {
    RailError::Refused("fixture tap lock poisoned".to_owned())
}

impl Harness {
    pub fn refuse(&self, subject: &str, words: &str) -> Result<(), RailError> {
        self.0
            .lock()
            .map_err(poisoned)?
            .refused
            .insert(subject.to_owned(), words.to_owned());
        Ok(())
    }

    pub fn open(&self, channel: &str) -> Result<Arc<Slot>, RailError> {
        let mut state = self.0.lock().map_err(poisoned)?;
        if let Some(words) = state.refused.get(channel) {
            return Err(RailError::Refused(words.clone()));
        }
        let slot = Arc::new(Slot::default());
        state.opened.push(channel.to_owned());
        state.history.push(Arc::clone(&slot));
        state.slots.insert(channel.to_owned(), Arc::clone(&slot));
        drop(state);
        Ok(slot)
    }

    pub fn opened(&self) -> Result<Vec<String>, RailError> {
        Ok(self.0.lock().map_err(poisoned)?.opened.clone())
    }

    pub fn active(&self) -> Result<usize, RailError> {
        let state = self.0.lock().map_err(poisoned)?;
        state.history.iter().try_fold(0, |count, slot| {
            Ok(count + usize::from(!slot.state.lock().map_err(poisoned)?.closed))
        })
    }

    pub fn wait_closed(&self, subject: &str) -> Result<(), RailError> {
        let slot = self
            .0
            .lock()
            .map_err(poisoned)?
            .slots
            .get(subject)
            .cloned()
            .ok_or_else(|| RailError::Refused(format!("no fixture subscription on {subject}")))?;
        let mut state = slot.state.lock().map_err(poisoned)?;
        while !state.closed {
            state = slot.changed.wait(state).map_err(poisoned)?;
        }
        drop(state);
        Ok(())
    }

    pub fn feed(&self, subject: &str, record: Record) -> Result<(), RailError> {
        let slot = self
            .0
            .lock()
            .map_err(poisoned)?
            .slots
            .get(subject)
            .cloned()
            .ok_or_else(|| RailError::Refused(format!("no fixture subscription on {subject}")))?;
        let mut state = slot.state.lock().map_err(poisoned)?;
        while state.record.is_some() && !state.closed {
            state = slot.changed.wait(state).map_err(poisoned)?;
        }
        if state.closed {
            return Err(RailError::Refused("fixture closed".to_owned()));
        }
        state.record = Some(record);
        slot.changed.notify_all();
        while state.record.is_some() && !state.closed {
            state = slot.changed.wait(state).map_err(poisoned)?;
        }
        drop(state);
        Ok(())
    }
}

impl super::TapReader for Arc<Slot> {
    fn receive(&self) -> Result<Value, RailError> {
        Slot::receive(self)
    }
}

impl super::TapCloser for Slot {
    fn close(&self) -> Result<(), RailError> {
        Self::close(self)
    }
}