vertep_runtime/
transport.rs

1//! Transports and address resolution.
2//!
3//! A [`Transport`] carries a message into a mailbox and is registered as an SCXML Event I/O
4//! Processor under its type IRI, which `<send type="...">` selects; the runtime lists every
5//! registered transport in `_ioprocessors`. M2 ships the in-process [`LocalTransport`].
6//!
7//! The [`Directory`] maps an address to the mailbox that serves it. Addresses hosted on this
8//! node are resolved first; any other address goes to the [`Resolver`] registered for its IRI
9//! scheme. Resolution depends only on the address.
10
11use std::collections::BTreeMap;
12use std::sync::{Arc, RwLock};
13
14use vertep_core::session::SCXML_EVENT_PROCESSOR;
15
16use crate::address::Address;
17use crate::mailbox::{Envelope, Mailbox};
18
19/// A delivery failure; the outbox keeps the send and retries it.
20#[derive(Clone, Debug, PartialEq, Eq)]
21pub struct TransportError(pub String);
22
23impl core::fmt::Display for TransportError {
24    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
25        write!(f, "transport error: {}", self.0)
26    }
27}
28
29impl std::error::Error for TransportError {}
30
31/// The transport contract: an SCXML Event I/O Processor that delivers into mailboxes.
32// @tg: implements:: [[openspec:messaging#Pluggable Transports]]
33pub trait Transport: Send + Sync {
34    /// The Event I/O Processor type IRI.
35    fn type_iri(&self) -> &str;
36    /// Other names `<send type>` may use for this transport.
37    fn aliases(&self) -> Vec<String> {
38        Vec::new()
39    }
40    /// The location listed in `_ioprocessors` for an actor at `address`.
41    fn location(&self, address: &Address) -> String {
42        address.as_str().to_owned()
43    }
44    /// Delivers `env` into `mailbox`.
45    ///
46    /// # Errors
47    ///
48    /// [`TransportError`] when the message was not accepted.
49    fn deliver(&self, mailbox: &dyn Mailbox, env: Envelope) -> Result<(), TransportError>;
50}
51
52/// The in-process transport: the SCXML Event I/O Processor
53/// (`http://www.w3.org/TR/scxml/#SCXMLEventProcessor`, alias `scxml`).
54#[derive(Clone, Copy, Debug, Default)]
55pub struct LocalTransport;
56
57impl Transport for LocalTransport {
58    fn type_iri(&self) -> &str {
59        SCXML_EVENT_PROCESSOR
60    }
61
62    fn aliases(&self) -> Vec<String> {
63        vec!["scxml".to_owned()]
64    }
65
66    fn deliver(&self, mailbox: &dyn Mailbox, env: Envelope) -> Result<(), TransportError> {
67        mailbox
68            .enqueue(env)
69            .map_err(|e| TransportError(e.to_string()))
70    }
71}
72
73/// Resolves addresses of one IRI scheme to mailboxes.
74pub trait Resolver: Send + Sync {
75    /// The mailbox serving `address`, if this resolver knows it.
76    fn resolve(&self, address: &Address) -> Option<Arc<dyn Mailbox>>;
77}
78
79/// An address the directory could not resolve.
80#[derive(Clone, Debug, PartialEq, Eq)]
81pub struct ResolveError(pub String);
82
83impl core::fmt::Display for ResolveError {
84    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
85        write!(f, "cannot resolve {}", self.0)
86    }
87}
88
89impl std::error::Error for ResolveError {}
90
91/// The resolvers of non-local addresses, keyed by IRI scheme.
92// @tg: implements:: [[openspec:messaging#Address Resolution]]
93#[derive(Default)]
94pub struct Directory {
95    resolvers: RwLock<BTreeMap<String, Arc<dyn Resolver>>>,
96}
97
98impl std::fmt::Debug for Directory {
99    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
100        let schemes: Vec<String> = self
101            .resolvers
102            .read()
103            .map(|r| r.keys().cloned().collect())
104            .unwrap_or_default();
105        f.debug_struct("Directory")
106            .field("schemes", &schemes)
107            .finish()
108    }
109}
110
111impl Directory {
112    /// Registers `resolver` for addresses whose scheme is `scheme`.
113    pub fn register(&self, scheme: &str, resolver: Arc<dyn Resolver>) {
114        if let Ok(mut r) = self.resolvers.write() {
115            r.insert(scheme.to_ascii_lowercase(), resolver);
116        }
117    }
118
119    /// The mailbox of a non-local `address`, from the resolver of its scheme.
120    ///
121    /// # Errors
122    ///
123    /// [`ResolveError`] when no resolver knows the address.
124    pub fn resolve(&self, address: &Address) -> Result<Arc<dyn Mailbox>, ResolveError> {
125        let r = self
126            .resolvers
127            .read()
128            .ok()
129            .and_then(|r| r.get(&address.scheme().to_ascii_lowercase()).cloned());
130        r.and_then(|r| r.resolve(address))
131            .ok_or_else(|| ResolveError(address.as_str().to_owned()))
132    }
133}
134
135/// The registered transports, by type IRI and alias.
136#[derive(Clone, Default)]
137pub struct Transports {
138    by_type: BTreeMap<String, Arc<dyn Transport>>,
139    primary: Vec<Arc<dyn Transport>>,
140}
141
142impl std::fmt::Debug for Transports {
143    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
144        f.debug_struct("Transports")
145            .field("types", &self.by_type.keys().collect::<Vec<_>>())
146            .finish()
147    }
148}
149
150impl Transports {
151    /// Only the local transport.
152    pub fn local() -> Self {
153        let mut t = Transports::default();
154        t.register(Arc::new(LocalTransport));
155        t
156    }
157
158    /// Registers `t` under its type IRI and aliases.
159    pub fn register(&mut self, t: Arc<dyn Transport>) {
160        self.by_type.insert(t.type_iri().to_owned(), t.clone());
161        for a in t.aliases() {
162            self.by_type.insert(a, t.clone());
163        }
164        self.primary.push(t);
165    }
166
167    /// The transport for a send type.
168    pub fn get(&self, type_: &str) -> Option<&Arc<dyn Transport>> {
169        self.by_type.get(type_)
170    }
171
172    /// Every type name a `<send>` may use.
173    pub fn send_types(&self) -> Vec<String> {
174        self.by_type.keys().cloned().collect()
175    }
176
177    /// The `_ioprocessors` value for an actor at `address`: each transport's type IRI with its
178    /// location.
179    pub fn ioprocessors(&self, address: &Address) -> vertep_core::datamodel::Value {
180        use vertep_core::datamodel::Value;
181        let mut entries = Vec::new();
182        for (name, t) in &self.by_type {
183            entries.push((
184                name.clone(),
185                Value::Object(vec![(
186                    "location".to_owned(),
187                    Value::String(t.location(address)),
188                )]),
189            ));
190        }
191        Value::Object(entries)
192    }
193}
vertep_runtime/transport.rs

vertep_runtime/
transport.rs
