liminal-server 0.24.5

//! Every serving subject is a call line, including worker replacements and retirement.

use super::model::{ThisDevice, TopologyChange, TopologyModel};
use super::records::{Mount, canonical_bytes};
use super::wall::WallBase;
use std::time::Instant;

type TestResult = Result<(), Box<dyn std::error::Error>>;

fn model() -> TopologyModel {
    TopologyModel::new(
        ThisDevice {
            id: "local".to_owned(),
            liminal: "1".to_owned(),
            started: Instant::now(),
            depot: false,
        },
        [],
        [],
        [],
        [],
    )
}

fn document(model: &TopologyModel) -> Result<serde_json::Value, Box<dyn std::error::Error>> {
    let wall = WallBase::fixed(1_791_881_244_000, Instant::now());
    Ok(serde_json::from_slice(&canonical_bytes(
        &model.document(&wall).0,
        "topology",
    )?)?)
}

#[test]
fn service_line_responder_subject_has_exchange_and_serving_stop() -> TestResult {
    let mut model = model();
    assert!(model.apply(TopologyChange::Service {
        subject: "jobs.studio".to_owned(),
        present: true
    }));
    let value = document(&model)?;
    assert_eq!(
        value["lines"],
        serde_json::json!([{ "id": "jobs.studio", "kind": "call", "life": "passing", "state": "live", "stops": ["local", "jobs.studio"] }])
    );
    assert!(!model.apply(TopologyChange::Service {
        subject: "jobs.studio".to_owned(),
        present: true
    }));
    assert!(model.apply(TopologyChange::Service {
        subject: "jobs.studio".to_owned(),
        present: false
    }));
    assert_eq!(document(&model)?["lines"], serde_json::json!([]));
    Ok(())
}

#[test]
fn service_line_worker_replacement_and_close_remove_old_subjects() -> TestResult {
    let mut model = model();
    model.apply(TopologyChange::ConnectionOpened {
        id: 7,
        mount: Mount::Tcp,
        participant: None,
    });
    let worker = |subject: &str| TopologyChange::Worker {
        connection: 7,
        identity: "studio".to_owned(),
        subjects: vec![subject.to_owned()],
        activities: vec![],
    };
    model.apply(worker("jobs.studio"));
    assert_eq!(
        document(&model)?["lines"],
        serde_json::json!([{ "id": "jobs.studio", "kind": "call", "life": "passing", "state": "live", "stops": ["local", "studio"] }])
    );
    model.apply(worker("jobs.other"));
    assert_eq!(document(&model)?["lines"][0]["id"], "jobs.other");
    assert_eq!(document(&model)?["lines"].as_array().map(Vec::len), Some(1));
    model.apply(TopologyChange::ConnectionClosed { id: 7 });
    assert_eq!(document(&model)?["lines"], serde_json::json!([]));
    Ok(())
}

#[test]
fn service_line_boot_responder_is_a_line_without_registration_replay() -> TestResult {
    let model = TopologyModel::new(
        ThisDevice {
            id: "local".to_owned(),
            liminal: "1".to_owned(),
            started: Instant::now(),
            depot: false,
        },
        ["jobs.studio".to_owned()],
        [],
        [],
        [],
    );
    assert_eq!(document(&model)?["lines"][0]["id"], "jobs.studio");
    Ok(())
}

#[derive(Default)]
struct Published(std::sync::Mutex<Vec<serde_json::Value>>);
impl super::ServerPublisher for Published {
    fn publish_console(&self, _: &str, bytes: &[u8]) -> Result<(), super::ConsoleError> {
        self.0
            .lock()
            .map_err(|_| crate::tap::TapError::Poisoned {
                what: "fixture records",
            })?
            .push(
                serde_json::from_slice(bytes).map_err(|source| super::ConsoleError::Encode {
                    record: "fixture",
                    source,
                })?,
            );
        Ok(())
    }
    fn retire_console_channel(&self, _: &str) -> Result<(), super::ConsoleError> {
        Ok(())
    }
}

#[test]
fn service_line_retirement_publishes_line_remove_before_the_snapshot() -> TestResult {
    let hub = super::ConsoleHub::new(
        "local",
        model(),
        WallBase::fixed(1_791_881_244_000, Instant::now()),
    )?;
    let out = Published::default();
    hub.attach(super::TOPOLOGY_SUBJECT)?;
    for subject in ["jobs.studio", "tasks.studio"] {
        hub.emit(TopologyChange::Service {
            subject: subject.to_owned(),
            present: true,
        })?;
    }
    hub.pump(&out)?;
    out.0.lock().map_err(|_| "records")?.clear();
    for subject in ["jobs.studio", "tasks.studio"] {
        hub.emit(TopologyChange::Service {
            subject: subject.to_owned(),
            present: false,
        })?;
    }
    hub.pump(&out)?;
    let records = out.0.lock().map_err(|_| "records")?;
    let removals: Vec<_> = records
        .iter()
        .filter(|record| record["change"]["op"] == "line.remove")
        .collect();
    assert_eq!(
        removals.len(),
        2,
        "each retired service subject must tell the feed line.remove"
    );
    assert_eq!(removals[0]["change"]["id"], "jobs.studio");
    assert_eq!(removals[1]["change"]["id"], "tasks.studio");
    assert_eq!(
        removals[1]["rev"].as_u64(),
        removals[0]["rev"].as_u64().map(|rev| rev + 1)
    );
    assert_eq!(
        records.last().ok_or("snapshot")?["lines"],
        serde_json::json!([])
    );
    assert_eq!(records.len(), 3);
    assert_eq!(records[0]["type"], "topology.change");
    assert_eq!(records[1]["type"], "topology.change");
    assert_eq!(records[2]["type"], "topology");
    assert_eq!(
        records[2]["rev"].as_u64(),
        removals[1]["rev"].as_u64().map(|rev| rev + 1),
        "the whole document must follow the last removal's revision"
    );
    Ok(())
}