ocpi_kit/hub/
routing_table.rs

1//! Which platform hosts which party, and whether it is reachable.
2
3use std::collections::BTreeMap;
4use std::sync::RwLock;
5
6use crate::client::Peer;
7use crate::transport::{OcpiError, StatusCode};
8use crate::types::PartyRef;
9use crate::v2_3_0::hub_client_info::ConnectionStatus;
10use crate::v2_3_0::types::Role;
11use crate::{InterfaceRole, ModuleId};
12
13/// One platform connected to the hub, and the parties it speaks for.
14#[derive(Debug)]
15pub struct ConnectedPlatform {
16    /// A stable identifier for the platform.
17    pub platform_id: String,
18    /// How to call it.
19    pub peer: Peer,
20    /// The parties it hosts, with the role each fills.
21    pub parties: Vec<(PartyRef, Role)>,
22    /// Whether messages can currently be delivered to it.
23    pub status: ConnectionStatus,
24}
25
26impl ConnectedPlatform {
27    /// Whether this platform hosts `party`.
28    #[must_use]
29    pub fn hosts(&self, party: &PartyRef) -> bool {
30        self.parties.iter().any(|(p, _)| p == party)
31    }
32
33    /// The role `party` fills at this platform.
34    #[must_use]
35    pub fn role_of(&self, party: &PartyRef) -> Option<Role> {
36        self.parties.iter().find(|(p, _)| p == party).map(|(_, r)| *r)
37    }
38
39    /// Whether messages can currently be delivered.
40    #[must_use]
41    pub fn is_reachable(&self) -> bool {
42        self.status == ConnectionStatus::Connected
43    }
44
45    /// Whether this platform implements a module in a given interface role.
46    #[must_use]
47    pub fn implements(&self, module: &ModuleId, role: InterfaceRole) -> bool {
48        self.peer.implements(module, role)
49    }
50}
51
52/// The hub's map from party to platform.
53///
54/// A hub's whole job is answering "who is `NL/TNM`, and can I reach them right now?", and turning
55/// a "no" into the right one of the four `4xxx` codes:
56///
57/// | Code | Meaning |
58/// |---|---|
59/// | `4001` | Unknown receiver: the `OCPI-to-*` address is unknown |
60/// | `4002` | Timeout on a forwarded request |
61/// | `4003` | Connection problem: the receiving party is not connected |
62/// | `4000` | Anything else |
63///
64/// Spec: 2.3.0 §status_codes_4xxx_hub_errors
65#[derive(Debug, Default)]
66pub struct RoutingTable {
67    platforms: RwLock<BTreeMap<String, ConnectedPlatform>>,
68}
69
70impl RoutingTable {
71    /// An empty table.
72    #[must_use]
73    pub fn new() -> Self {
74        Self::default()
75    }
76
77    /// Adds or replaces a platform.
78    pub fn upsert(&self, platform: ConnectedPlatform) {
79        self.platforms
80            .write()
81            .expect("routing table lock poisoned")
82            .insert(platform.platform_id.clone(), platform);
83    }
84
85    /// Removes a platform, as an unregistration does.
86    pub fn remove(&self, platform_id: &str) -> bool {
87        self.platforms.write().expect("routing table lock poisoned").remove(platform_id).is_some()
88    }
89
90    /// Records a platform's connection status, which the `hubclientinfo` module publishes.
91    pub fn set_status(&self, platform_id: &str, status: ConnectionStatus) -> bool {
92        let mut platforms = self.platforms.write().expect("routing table lock poisoned");
93        match platforms.get_mut(platform_id) {
94            Some(platform) => {
95                platform.status = status;
96                true
97            }
98            None => false,
99        }
100    }
101
102    /// Runs `f` over the platform hosting `party`.
103    ///
104    /// # Errors
105    ///
106    /// Returns `4001 Unknown receiver` when no platform hosts the party, and `4003 Connection
107    /// problem` when the platform is known but not connected.
108    pub fn with_platform<T>(
109        &self,
110        party: &PartyRef,
111        f: impl FnOnce(&ConnectedPlatform) -> T,
112    ) -> Result<T, OcpiError> {
113        let platforms = self.platforms.read().expect("routing table lock poisoned");
114        let platform = platforms.values().find(|p| p.hosts(party)).ok_or_else(|| OcpiError::Remote {
115            status_code: StatusCode::UNKNOWN_RECEIVER,
116            status_message: Some(format!("the hub does not know {party}")),
117        })?;
118        if !platform.is_reachable() {
119            return Err(OcpiError::Remote {
120                status_code: StatusCode::CONNECTION_PROBLEM,
121                status_message: Some(format!("{party} is {}", platform.status)),
122            });
123        }
124        Ok(f(platform))
125    }
126
127    /// Whether the hub knows `party` at all, connected or not.
128    #[must_use]
129    pub fn knows(&self, party: &PartyRef) -> bool {
130        self.platforms.read().expect("routing table lock poisoned").values().any(|p| p.hosts(party))
131    }
132
133    /// The platform id hosting `party`.
134    #[must_use]
135    pub fn platform_of(&self, party: &PartyRef) -> Option<String> {
136        self.platforms
137            .read()
138            .expect("routing table lock poisoned")
139            .values()
140            .find(|p| p.hosts(party))
141            .map(|p| p.platform_id.clone())
142    }
143
144    /// The parties a Broadcast Push from `sender` should reach for `module`.
145    ///
146    /// > *For simplicity, connected clients might push (POST, PUT, PATCH) information to all
147    /// > connected clients with an "opposite role" … When using Broadcast Push, the Hub broadcasts
148    /// > received information to all connected clients … using its own party-id and country-code
149    /// > in the 'OCPI-from-' headers.*
150    ///
151    /// A party is a recipient when it is connected, fills a role that receives from the sender's
152    /// role, implements the module's Receiver interface, and is not the sender itself.
153    ///
154    /// Spec: 2.3.0 §transport_and_format_message_routing_broadcast_push
155    #[must_use]
156    pub fn broadcast_targets(
157        &self,
158        sender: &PartyRef,
159        sender_role: Role,
160        module: &ModuleId,
161    ) -> Vec<(String, PartyRef)> {
162        let platforms = self.platforms.read().expect("routing table lock poisoned");
163        let mut targets = Vec::new();
164        for platform in platforms.values() {
165            if !platform.is_reachable() || !platform.implements(module, InterfaceRole::Receiver) {
166                continue;
167            }
168            for (party, role) in &platform.parties {
169                if party == sender {
170                    continue;
171                }
172                if role.receives_broadcast_from(sender_role) {
173                    targets.push((platform.platform_id.clone(), party.clone()));
174                }
175            }
176        }
177        targets
178    }
179
180    /// Every party that implements a module's Sender interface, for a GET All.
181    ///
182    /// > *A client (Receiver) can request a GET on the Sender interface of a module implemented by
183    /// > a Hub. The Hub can then combine objects from different connected parties.*
184    ///
185    /// Spec: 2.3.0 §transport_and_format_get_all_via_hubs
186    #[must_use]
187    pub fn get_all_sources(&self, requester: &PartyRef, module: &ModuleId) -> Vec<(String, PartyRef)> {
188        let platforms = self.platforms.read().expect("routing table lock poisoned");
189        let mut sources = Vec::new();
190        for platform in platforms.values() {
191            if !platform.is_reachable() || !platform.implements(module, InterfaceRole::Sender) {
192                continue;
193            }
194            for (party, _) in &platform.parties {
195                if party != requester {
196                    sources.push((platform.platform_id.clone(), party.clone()));
197                }
198            }
199        }
200        sources
201    }
202
203    /// Every platform the hub knows, connected or not.
204    #[must_use]
205    pub fn platform_ids(&self) -> Vec<String> {
206        self.platforms.read().expect("routing table lock poisoned").keys().cloned().collect()
207    }
208
209    /// Every party the hub knows, with its role and connection status.
210    ///
211    /// This is what the `hubclientinfo` module publishes.
212    #[must_use]
213    pub fn client_info(&self) -> Vec<(PartyRef, Role, ConnectionStatus)> {
214        let platforms = self.platforms.read().expect("routing table lock poisoned");
215        platforms
216            .values()
217            .flat_map(|p| p.parties.iter().map(move |(party, role)| (party.clone(), *role, p.status)))
218            .collect()
219    }
220}
221
222#[cfg(test)]
223mod tests {
224    use super::*;
225    use crate::VersionNumber;
226    use crate::transport::CredentialsToken;
227    use crate::types::Url;
228
229    fn platform(
230        id: &str,
231        parties: &[(&str, &str, Role)],
232        status: ConnectionStatus,
233        modules: &[(ModuleId, InterfaceRole)],
234    ) -> ConnectedPlatform {
235        let mut builder = Peer::builder(VersionNumber::V2_3_0, CredentialsToken::new("t").unwrap());
236        for (module, role) in modules {
237            builder = builder.endpoint(
238                module.clone(),
239                *role,
240                Url::new(format!("https://{id}.example.com/{module}")).unwrap(),
241            );
242        }
243        ConnectedPlatform {
244            platform_id: id.to_owned(),
245            peer: builder.build(),
246            parties: parties
247                .iter()
248                .map(|(cc, pid, role)| (PartyRef::new(*cc, *pid).unwrap(), *role))
249                .collect(),
250            status,
251        }
252    }
253
254    fn table() -> RoutingTable {
255        let t = RoutingTable::new();
256        t.upsert(platform(
257            "cpo",
258            &[("NL", "TNM", Role::Cpo)],
259            ConnectionStatus::Connected,
260            &[(ModuleId::Locations, InterfaceRole::Sender), (ModuleId::Tokens, InterfaceRole::Receiver)],
261        ));
262        t.upsert(platform(
263            "msp",
264            &[("DE", "ABC", Role::Emsp), ("DE", "XYZ", Role::Nsp)],
265            ConnectionStatus::Connected,
266            &[(ModuleId::Locations, InterfaceRole::Receiver), (ModuleId::Tokens, InterfaceRole::Sender)],
267        ));
268        t.upsert(platform(
269            "gone",
270            &[("FR", "OLD", Role::Emsp)],
271            ConnectionStatus::Offline,
272            &[(ModuleId::Locations, InterfaceRole::Receiver)],
273        ));
274        t
275    }
276
277    #[test]
278    fn an_unknown_party_is_4001_and_a_disconnected_one_is_4003() {
279        let t = table();
280        assert!(t.with_platform(&PartyRef::new("NL", "TNM").unwrap(), |p| p.platform_id.clone()).is_ok());
281
282        let unknown = t.with_platform(&PartyRef::new("XX", "NON").unwrap(), |_| ()).unwrap_err();
283        assert_eq!(unknown.status_code(), StatusCode::UNKNOWN_RECEIVER);
284
285        let offline = t.with_platform(&PartyRef::new("FR", "OLD").unwrap(), |_| ()).unwrap_err();
286        assert_eq!(offline.status_code(), StatusCode::CONNECTION_PROBLEM);
287        assert!(t.knows(&PartyRef::new("FR", "OLD").unwrap()), "known, just not reachable");
288    }
289
290    #[test]
291    fn a_broadcast_from_a_cpo_reaches_the_emsp_like_roles_that_implement_the_module() {
292        let t = table();
293        let targets =
294            t.broadcast_targets(&PartyRef::new("NL", "TNM").unwrap(), Role::Cpo, &ModuleId::Locations);
295        let parties: Vec<String> = targets.iter().map(|(_, p)| p.to_string()).collect();
296        assert!(parties.contains(&"DE/ABC".to_owned()), "{parties:?}");
297        assert!(parties.contains(&"DE/XYZ".to_owned()), "an NSP receives Locations too");
298        assert!(!parties.contains(&"FR/OLD".to_owned()), "an offline platform is skipped");
299        assert!(!parties.contains(&"NL/TNM".to_owned()), "the sender is not a target");
300    }
301
302    #[test]
303    fn a_broadcast_skips_platforms_that_do_not_implement_the_module() {
304        let t = table();
305        // The CPO platform implements Tokens/Receiver, so a broadcast of Tokens from the eMSP
306        // reaches it; the other eMSP-like party does not receive from an eMSP.
307        let targets =
308            t.broadcast_targets(&PartyRef::new("DE", "ABC").unwrap(), Role::Emsp, &ModuleId::Tokens);
309        assert_eq!(targets.len(), 1);
310        assert_eq!(targets[0].1, PartyRef::new("NL", "TNM").unwrap());
311    }
312
313    #[test]
314    fn get_all_collects_every_sender_but_the_requester() {
315        let t = table();
316        let sources = t.get_all_sources(&PartyRef::new("DE", "ABC").unwrap(), &ModuleId::Locations);
317        assert_eq!(sources.len(), 1);
318        assert_eq!(sources[0].1, PartyRef::new("NL", "TNM").unwrap());
319    }
320
321    #[test]
322    fn status_changes_take_effect_immediately() {
323        let t = table();
324        assert!(t.set_status("cpo", ConnectionStatus::Offline));
325        assert!(!t.set_status("nope", ConnectionStatus::Offline));
326        let err = t.with_platform(&PartyRef::new("NL", "TNM").unwrap(), |_| ()).unwrap_err();
327        assert_eq!(err.status_code(), StatusCode::CONNECTION_PROBLEM);
328        assert_eq!(t.client_info().len(), 4);
329    }
330}
ocpi_kit/hub/routing_table.rs

ocpi_kit/hub/
routing_table.rs
