liminal-cli 0.3.4

//! A rail answering from a fixture: the topology record it is given, and
//! for each subject a reply or a refusal in fixed words. Every call is
//! counted exactly as the live rail counts it.

use std::collections::BTreeMap;
use std::sync::Mutex;

use serde_json::{Value, json};

use super::{Counts, Rail, RailError};

/// The fixture's pass for an audience is this prefix and the audience.
pub const FAKE_PASS_PREFIX: &str = "fixture-pass-for-";

/// What a fixture subject answers.
#[derive(Clone, Debug)]
pub enum Answer {
    /// `{"echoed": <input>}`.
    Echo,
    /// A fixed reply.
    Reply(Value),
    /// The server's refusal words.
    Refuse(String),
}

/// The fixture rail.
#[derive(Debug)]
pub struct Fake {
    topology: Mutex<Result<Value, RailError>>,
    subjects: BTreeMap<String, Answer>,
    pushed: Mutex<Vec<(String, Value)>>,
    counts: Counts,
    request_bytes: usize,
    flow_steps: usize,
    pub taps: super::fake_taps::Harness,
}

impl Fake {
    /// A rail serving `record` on the topology and `subjects` on request.
    pub fn new(record: Value, subjects: &[(&str, Answer)]) -> Self {
        Self {
            topology: Mutex::new(Ok(record)),
            subjects: subjects
                .iter()
                .map(|(subject, answer)| ((*subject).to_owned(), answer.clone()))
                .collect(),
            pushed: Mutex::new(Vec::new()),
            counts: Counts::default(),
            request_bytes: 4096,
            flow_steps: 4,
            taps: super::fake_taps::Harness::default(),
        }
    }

    /// The topology from now on.
    pub fn set_topology(&self, record: Result<Value, RailError>) {
        if let Ok(mut topology) = self.topology.lock() {
            *topology = record;
        }
    }

    /// Every request pushed so far, in order, with its bytes as JSON.
    pub fn pushed(&self) -> Vec<(String, Value)> {
        self.pushed.lock().map(|p| p.clone()).unwrap_or_default()
    }

    /// A record with one worker station on `subject` carrying `activities`.
    pub fn record(rev: u64, subject: &str, activities: &Value) -> Value {
        json!({
            "v": 1,
            "type": "topology",
            "rev": rev,
            "this_device": "mac",
            "devices": [{ "id": "mac", "name": "mac", "kind": "device" }],
            "stations": [
                { "id": "mac", "name": "mac", "device": "mac", "kind": "device", "state": "live" },
                {
                    "id": subject,
                    "name": "echo-worker",
                    "device": "mac",
                    "kind": "worker",
                    "state": "live",
                    "subjects": [subject],
                    "activities": activities
                }
            ],
            "lines": [],
            "links": [],
            "connections": []
        })
    }
}

impl Rail for Fake {
    fn topology(&self) -> Result<Value, RailError> {
        Counts::add(&self.counts.connections_opened);
        Counts::add(&self.counts.subscribes);
        let record = self
            .topology
            .lock()
            .map_err(|_| RailError::Refused("the fixture's lock is poisoned".to_owned()))?
            .clone();
        Counts::add(&self.counts.connections_closed);
        if record.is_ok() {
            Counts::add(&self.counts.records);
        }
        record
    }

    fn request(
        &self,
        subject: &str,
        input: &Value,
        reply: bool,
    ) -> Result<Option<Value>, RailError> {
        Counts::add(&self.counts.connections_opened);
        Counts::add(&self.counts.requests);
        let outcome = self.subjects.get(subject).map_or_else(
            || Err(RailError::Refused(format!("no station serves {subject}"))),
            |answer| {
                if let Ok(mut pushed) = self.pushed.lock() {
                    pushed.push((subject.to_owned(), input.clone()));
                }
                match answer {
                    Answer::Refuse(words) => Err(RailError::Refused(words.clone())),
                    _ if !reply => Ok(None),
                    Answer::Echo => Ok(Some(json!({ "echoed": input }))),
                    Answer::Reply(value) => Ok(Some(value.clone())),
                }
            },
        );
        if matches!(outcome, Ok(Some(_))) {
            Counts::add(&self.counts.replies);
        }
        Counts::add(&self.counts.connections_closed);
        outcome
    }

    fn pass_for(&self, audience: &str) -> Result<String, RailError> {
        Ok(format!("{FAKE_PASS_PREFIX}{audience}"))
    }

    fn tap(&self, channel: &str) -> Result<super::TapSubscription, RailError> {
        let slot = self.taps.open(channel)?;
        Ok(super::TapSubscription {
            reader: Box::new(std::sync::Arc::clone(&slot)),
            closer: slot,
        })
    }

    fn server_address(&self) -> &'static str {
        "fixture:7000"
    }

    fn counts(&self) -> &Counts {
        &self.counts
    }

    fn request_bytes(&self) -> usize {
        self.request_bytes
    }

    fn flow_steps(&self) -> usize {
        self.flow_steps
    }
}