liminal-server 0.25.4

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
112
113
114
115
116

//! One configured chooser supplies the bridge's carrier and the device's incoming endpoint.
use super::super::link_udp_endpoint::UdpEndpoint;
use super::{ServerError, refusal};
use crate::config::types::{MachineConfig, OutboundLinkConfig};
use crate::link::LinkHost;
use crate::link_carrier::choice::ChoiceLink;
use crate::link_open::OpenError;
use crate::link_open::network::{Network, NetworkBackend};
use crate::server::ConnectionSupervisor;
use crate::server::connection::LiminalConnectionServices;
use std::sync::Arc;

pub(super) enum ChoiceOutcome {
    Opened(Box<ChoiceLink>),
    Held { grant: String, draft: String },
    Refused { reason: String },
    DialRefused { reason: String },
}
pub(super) struct ConfiguredChooser {
    network: Arc<Network>,
    machine: MachineConfig,
    services: Option<Arc<LiminalConnectionServices>>,
    supervisor: Option<ConnectionSupervisor>,
    endpoint: Option<UdpEndpoint>,
}
impl ConfiguredChooser {
    pub(super) fn new(host: &Arc<LinkHost>, machine: &MachineConfig) -> Self {
        Self {
            network: Network::new(
                host.clone(),
                machine.clone(),
                machine.carrier_choice.clone(),
            ),
            machine: machine.clone(),
            services: None,
            supervisor: None,
            endpoint: None,
        }
    }
    pub(super) fn with_topology(mut self, services: Arc<LiminalConnectionServices>) -> Self {
        self.services = Some(services);
        self
    }
    pub(super) fn with_endpoint(
        mut self,
        supervisor: &ConnectionSupervisor,
    ) -> Result<Self, ServerError> {
        self.supervisor = Some(supervisor.clone());
        if self.machine.links.is_empty() {
            self.serve_incoming().map_err(|error| refusal(error.0))?;
        }
        Ok(self)
    }
    fn serve_incoming(&mut self) -> Result<(), OpenError> {
        let choice = &self.machine.carrier_choice;
        if self.endpoint.is_some()
            || choice.broker_address.is_none()
            || choice.broker_echo.is_none()
        {
            return Ok(());
        }
        self.endpoint = Some(UdpEndpoint::start(
            self.network.clone(),
            self.supervisor
                .as_ref()
                .ok_or_else(|| OpenError("chooser_has_no_supervisor".to_owned()))?
                .clone(),
            self.services
                .as_ref()
                .ok_or_else(|| OpenError("chooser_has_no_services".to_owned()))?
                .clone(),
        )?);
        Ok(())
    }
    pub(super) fn open(&mut self, entry: &OutboundLinkConfig) -> Result<ChoiceOutcome, OpenError> {
        let services = self
            .services
            .as_ref()
            .ok_or_else(|| OpenError("chooser_has_no_services".to_owned()))?;
        let backend = NetworkBackend::new(self.network.clone(), services.clone(), entry)?;
        match crate::link_open::open(entry.clone(), self.machine.carrier_choice.clone(), backend) {
            Ok(link) => {
                if link.carrier() == Some(crate::link_carrier::Carrier::Direct) {
                    self.serve_incoming()?;
                }
                Ok(ChoiceOutcome::Opened(Box::new(ChoiceLink::chosen(link)?)))
            }
            Err(error) if error.0.starts_with("link_held:") => {
                let (grant, draft) = error
                    .0
                    .strip_prefix("link_held:")
                    .and_then(|held| held.split_once(':'))
                    .ok_or_else(|| OpenError("invalid_held_link_answer".to_owned()))?;
                Ok(ChoiceOutcome::Held {
                    grant: grant.to_owned(),
                    draft: draft.to_owned(),
                })
            }
            Err(error) if error.0.starts_with("link_refused:") => {
                Ok(ChoiceOutcome::Refused { reason: error.0 })
            }
            Err(error) if error.0.starts_with("all_link_carriers_refused:") => {
                Ok(ChoiceOutcome::DialRefused { reason: error.0 })
            }
            Err(error) => Err(error),
        }
    }
    pub(super) fn stop_handle(&self) -> crate::link_open::Stop {
        self.network.stop_handle()
    }
    pub(super) fn finish(&mut self) -> Result<(), ServerError> {
        self.endpoint.take().map_or(Ok(()), |mut endpoint| {
            endpoint.finish().map_err(|error| refusal(error.0))
        })
    }
}