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
117
118
119
120
121

//! Replacement bridges resume the same subscriptions and cumulative observations.
use super::{ExportSubscriptions, LinkBridge, Observation, ServerError, failure, run_worker};
use crate::link_carrier::choice::ChoiceLink as DialedLink;
use crate::server::connection::services::LiminalConnectionServices;
use std::sync::Arc;

impl LinkBridge {
    pub(crate) fn start_named(
        link: impl Into<DialedLink>,
        services: Arc<LiminalConnectionServices>,
        peer: String,
    ) -> Result<Self, ServerError> {
        Self::start_resuming(link, services, peer, super::BridgeCounters::default())
    }

    pub(crate) fn start_resuming(
        link: impl Into<DialedLink>,
        services: Arc<LiminalConnectionServices>,
        peer: String,
        counters: super::BridgeCounters,
    ) -> Result<Self, ServerError> {
        Self::start_observing(link, services, peer, counters, |_| Ok(()))
    }

    pub(super) fn start_observing(
        link: impl Into<DialedLink>,
        services: Arc<LiminalConnectionServices>,
        peer: String,
        counters: super::BridgeCounters,
        prepare: impl FnOnce(&Observation) -> Result<(), ServerError>,
    ) -> Result<Self, ServerError> {
        let mut link = link.into();
        let interrupt = link
            .carrier
            .interrupt_handle()
            .map_err(|error| failure(error.to_string()))?;
        let observation = Observation::resume(
            interrupt
                .try_clone()
                .map_err(|error| failure(error.to_string()))?,
            counters,
        )?;
        prepare(&observation)?;
        let wake = Arc::clone(&observation.wake);
        let feeds = Arc::new(ExportSubscriptions::new(
            &link.link_id,
            &link.exports,
            Arc::clone(&wake),
            Arc::clone(&observation),
        )?);
        services.install_exporter(&feeds)?;
        crate::server::self_streams::cost::install_bridge_reader(&services, &feeds)?;
        observation.install_life(&link, &services)?;
        observation.watch_end(&wake)?;
        let worker_wake = Arc::clone(&wake);
        let worker_observation = Arc::clone(&observation);
        let link_id = link.link_id.clone();
        let dispatcher = tracing::dispatcher::get_default(Clone::clone);
        let worker = std::thread::Builder::new()
            .name("liminal-link-bridge".to_owned())
            .spawn(move || {
                tracing::dispatcher::with_default(&dispatcher, || {
                    run_worker(
                        &mut link,
                        &feeds,
                        &worker_wake,
                        &worker_observation,
                        &services,
                        &peer,
                    )
                })
            })
            .map_err(|error| failure(format!("link bridge worker: {error}")))?;
        Ok(Self {
            link_id,
            observation,
            wake,
            interrupt,
            worker: Some(worker),
        })
    }

    pub(crate) fn notify_end(
        &self,
        notify: Arc<dyn Fn() + Send + Sync>,
    ) -> Result<(), ServerError> {
        let wake = Arc::downgrade(&self.wake);
        self.observation
            .life
            .get()
            .ok_or_else(|| failure("bridge lifecycle absent"))?
            .notify_with(Arc::new(move || {
                if let Some(wake) = wake.upgrade() {
                    wake.notify();
                }
                notify();
            }))
            .map_err(failure)
    }
    pub(crate) fn end_reason(&self) -> Result<Option<String>, ServerError> {
        self.observation.life_reason().map_err(failure)
    }
    pub(crate) fn request_shutdown(&self) -> Result<(), ServerError> {
        self.wake.stop();
        self.observation.end_life(crate::link::LinkEnd::Shutdown)?;
        self.interrupt
            .shutdown(std::net::Shutdown::Both)
            .or_else(|error| {
                if error.kind() == std::io::ErrorKind::NotConnected {
                    Ok(())
                } else {
                    Err(error)
                }
            })
            .map_err(|error| failure(error.to_string()))
    }
    pub(crate) fn finish_retaining(mut self) -> Result<super::BridgeCounters, ServerError> {
        self.stop_worker()?;
        Ok(self.counters())
    }
}