logicaffeine_system/
net.rs

1//! `Net` — the cross-target relay-backed network handle the interpreter holds.
2//!
3//! One API, two transports selected at compile time: native dials with the async
4//! [`crate::relay::RelayClient`]; the browser dials with the callback-based
5//! `crate::relay_browser::RelayBrowserClient`. Both speak the same `RelayFrame`
6//! protocol to the same relay, so the interpreter's `Sync`/`Connect` lowering is
7//! written once and runs on both targets. This is the platform-capability seam
8//! for networking, the analogue of `get_platform_vfs()` for files.
9//!
10//! Events are drained, not awaited: the interpreter's `Sync` is a *sync point*
11//! (publish local state, then merge whatever has arrived) — [`Net::drain`] returns
12//! the pending messages without blocking, which keeps the tree-walker's linear
13//! execution model intact (no background task mutating the environment).
14
15#![cfg(any(all(not(target_arch = "wasm32"), feature = "relay"), target_arch = "wasm32"))]
16
17#[cfg(all(not(target_arch = "wasm32"), feature = "relay"))]
18mod imp {
19    use crate::relay::RelayClient;
20
21    /// Native relay handle (async `tokio-tungstenite` client).
22    pub struct Net {
23        client: RelayClient,
24    }
25
26    impl Net {
27        pub async fn connect(url: &str) -> Result<Net, String> {
28            Ok(Net { client: RelayClient::connect(url).await? })
29        }
30
31        pub async fn subscribe(&mut self, topic: &str) -> Result<(), String> {
32            self.client.subscribe(topic).await
33        }
34
35        pub fn publish(&self, topic: &str, data: Vec<u8>) -> Result<(), String> {
36            self.client.publish(topic, data)
37        }
38
39        /// Non-blocking: every message delivered since the last drain.
40        pub fn drain(&mut self) -> Vec<(String, Vec<u8>)> {
41            self.client.drain_events()
42        }
43    }
44}
45
46#[cfg(target_arch = "wasm32")]
47mod imp {
48    use std::cell::RefCell;
49    use std::collections::VecDeque;
50    use std::rc::Rc;
51
52    use crate::relay_browser::RelayBrowserClient;
53
54    /// Browser relay handle (`web-sys` WebSocket). Received events are pushed onto
55    /// a queue by the socket callback; `drain` empties it.
56    pub struct Net {
57        client: RelayBrowserClient,
58        events: Rc<RefCell<VecDeque<(String, Vec<u8>)>>>,
59    }
60
61    impl Net {
62        pub async fn connect(url: &str) -> Result<Net, String> {
63            let (open_tx, open_rx) = futures::channel::oneshot::channel::<()>();
64            let open_tx = Rc::new(RefCell::new(Some(open_tx)));
65            let events = Rc::new(RefCell::new(VecDeque::new()));
66
67            let opener = open_tx.clone();
68            let queue = events.clone();
69            let client = RelayBrowserClient::connect(
70                url,
71                move || {
72                    if let Some(tx) = opener.borrow_mut().take() {
73                        let _ = tx.send(());
74                    }
75                },
76                move |topic, data| queue.borrow_mut().push_back((topic, data)),
77            )
78            .map_err(|e| format!("relay connect failed: {e:?}"))?;
79
80            open_rx.await.map_err(|_| "socket closed before it opened".to_string())?;
81            Ok(Net { client, events })
82        }
83
84        pub async fn subscribe(&mut self, topic: &str) -> Result<(), String> {
85            // The browser socket is ordered, so a following publish cannot race
86            // the subscribe; no SubAck round-trip is needed.
87            self.client.subscribe(topic).map_err(|e| format!("subscribe failed: {e:?}"))
88        }
89
90        pub fn publish(&self, topic: &str, data: Vec<u8>) -> Result<(), String> {
91            self.client.publish(topic, data).map_err(|e| format!("publish failed: {e:?}"))
92        }
93
94        pub fn drain(&mut self) -> Vec<(String, Vec<u8>)> {
95            self.events.borrow_mut().drain(..).collect()
96        }
97    }
98}
99
100pub use imp::Net;
logicaffeine_system/net.rs

logicaffeine_system/
net.rs
