liminal-server 0.23.1

Standalone server for the liminal messaging bus
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111

//! Starts configured exporting bridges after the listeners bind and stops them before the store flush.
//! A refused dial leaves the named link down and is retried while the server keeps serving.

use crate::ServerError;
use crate::config::ServerConfig;
use crate::config::types::MachineConfig;
use crate::link::LinkHost;
use crate::link_bridge::BridgeRegistry;
use crate::server::connection::LysInstall;
use crate::server::connection::services::LiminalConnectionServices;
use std::sync::Arc;
#[path = "runtime_links/carriers.rs"]
mod carriers;
#[path = "runtime_links/retry.rs"]
mod retry;
use retry::RetryLinks;
#[cfg(feature = "test-support")]
#[path = "runtime_links/test_support.rs"]
pub mod test_support;

#[derive(Debug)]
struct Configuration {
    machine: MachineConfig,
    host: Arc<LinkHost>,
    services: Arc<LiminalConnectionServices>,
}

#[derive(Debug, Default)]
pub(in crate::server) struct ConfiguredLinks {
    configuration: Option<Configuration>,
    retry: Option<RetryLinks>,
}

pub(super) fn prepare(
    config: &ServerConfig,
    install: Option<&LysInstall>,
    services: &Arc<LiminalConnectionServices>,
) -> Result<ConfiguredLinks, ServerError> {
    let Some(machine) = config
        .auth
        .as_ref()
        .and_then(|auth| auth.lys.as_ref())
        .and_then(|lys| lys.machine.as_ref())
        .filter(|machine| {
            !machine.links.is_empty() || machine.carrier_choice.broker_echo.is_some()
        })
    else {
        return Ok(ConfiguredLinks::default());
    };
    let install =
        install.ok_or_else(|| refusal("configured outbound links have no installed Lys"))?;
    let Some(host) = install.link_host.as_ref() else {
        // The pass source names its boot refusal; serving does not require a link host.
        return Ok(ConfiguredLinks::default());
    };
    Ok(ConfiguredLinks {
        configuration: Some(Configuration {
            machine: machine.clone(),
            host: Arc::clone(host),
            services: Arc::clone(services),
        }),
        retry: None,
    })
}

impl ConfiguredLinks {
    pub(super) fn start(
        &mut self,
        registry: &BridgeRegistry,
        supervisor: &crate::server::ConnectionSupervisor,
    ) -> Result<(), ServerError> {
        let Some(configuration) = &self.configuration else {
            return Ok(());
        };
        self.retry = Some(RetryLinks::start(configuration, registry, supervisor)?);
        Ok(())
    }
    pub(super) fn finish(&mut self) -> Result<(), ServerError> {
        self.retry.take().map_or(Ok(()), |mut retry| retry.finish())
    }
}

fn refusal(message: impl Into<String>) -> ServerError {
    ServerError::ConfigValidation {
        message: message.into(),
    }
}

pub(super) fn lys(
    config: &ServerConfig,
) -> Result<(Option<tokio::runtime::Runtime>, Option<LysInstall>), ServerError> {
    let runtime = crate::server::connection::lys_ask_runtime(config.auth.as_ref())?;
    let install = super::lys_for(config, runtime.as_ref())?;
    Ok((runtime, install))
}

pub(in crate::server) fn shutdown(
    supervisor: &crate::server::ConnectionSupervisor,
    bound: std::time::Duration,
    stop_accepting: impl FnOnce() -> Result<(), ServerError>,
    publishers: (
        &mut ConfiguredLinks,
        impl FnOnce() -> Result<(), ServerError>,
    ),
) -> Result<(), ServerError> {
    let (links, finalize) = publishers;
    let links_shutdown = links.finish();
    let shutdown =
        crate::server::shutdown::sequence::run(supervisor, bound, stop_accepting, finalize);
    super::raise_in_precedence(links_shutdown, shutdown)
}