liminal-server 0.22.0

//! The configured backend uses the server's existing admitted wire and publish receipt.

use super::wire::federation::{Federation, LiveOpen};
use super::{Carrier, ChoiceConfig, TestResult, open};
use liminal::protocol::{CausalContext, MessageEnvelope};
use liminal_server::link_open::network::{Network, NetworkBackend};
use liminal_server::server::connection::services::LiminalConnectionServices;
use std::sync::Arc;

#[test]
fn the_configured_backend_crosses_a_real_publish_over_the_wire() -> TestResult {
    let federation = Federation::start(LiveOpen::Outright)?;
    let mut subscriber = federation.importer.subscribe("orders.created")?;
    let config = federation.outbound();
    let services = Arc::new(LiminalConnectionServices::empty()?);
    let choice = ChoiceConfig::default();
    let network = Network::new(
        Arc::clone(&federation.exporter.host),
        federation.exporter.config.clone(),
        choice.clone(),
    );
    let backend = NetworkBackend::new(Arc::clone(&network), Arc::clone(&services), &config)?;
    let mut link = open(config, choice, backend)?;
    assert_eq!(link.carrier(), Some(Carrier::Wire));
    let envelope = MessageEnvelope::new(
        subscriber.selected_schema(),
        CausalContext::independent(),
        br#"{"message":"confirmed importing publish"}"#.to_vec(),
    );
    federation
        .exporter
        .publish("orders.created", envelope.clone())?;
    let mut crossing = 1_u64.to_be_bytes().to_vec();
    crossing.extend(envelope.to_wire_bytes()?);
    link.send("orders.created", &crossing)?;
    assert_eq!(subscriber.receive()?.payload, envelope.payload);
    assert_eq!(federation.live_asks(), 1);
    link.interrupt_handle()?()?;
    drop(link);
    drop(network);
    services.conversation_supervisor().shutdown();
    services.channel_cluster().supervisor().shutdown();
    drop(services);
    federation.shutdown()?;
    Ok(())
}

#[test]
fn held_and_refused_opens_do_not_bypass_the_wire_answer() -> TestResult {
    for (answer, prefix) in [
        (LiveOpen::Held, "link_held:"),
        (LiveOpen::Refused, "link_refused:"),
    ] {
        let federation = Federation::start(answer)?;
        let config = federation.outbound();
        let services = Arc::new(LiminalConnectionServices::empty()?);
        let choice = ChoiceConfig::default();
        let network = Network::new(
            Arc::clone(&federation.exporter.host),
            federation.exporter.config.clone(),
            choice.clone(),
        );
        let backend = NetworkBackend::new(Arc::clone(&network), Arc::clone(&services), &config)?;
        let error = open(config, choice, backend)
            .err()
            .ok_or("deliberate open answer must refuse")?;
        assert!(error.0.starts_with(prefix));
        assert_eq!(federation.live_asks(), 1);
        drop(network);
        services.conversation_supervisor().shutdown();
        services.channel_cluster().supervisor().shutdown();
        drop(services);
        federation.shutdown()?;
    }
    Ok(())
}