umsh_mobile_core/
virtual_radio.rs

1//! The LAN virtual radio: a companion radio with no hardware, whose PHY is
2//! the LAN bearer's multicast group.
3//!
4//! The device is the production ULCP session on a file-backed platform.
5//! The platform attaches to it with the same HDLC-Lite bytes it would
6//! send a radio over a serial link, so everything above—the ULCP session,
7//! the mesh, the chat—is the code that runs against a T1000-E. The
8//! device runs on a worker thread of its own for as long as this object
9//! lives, attached or not: detached, it acknowledges for its host and
10//! queues, and reports each queued frame so the platform can tell the
11//! user.
12//!
13//! The group is the device's own `PROP_LAN_BEARER_GROUP`, set through
14//! the same management as any radio's, and a network to join it on comes
15//! and goes as `PROP_LAN_BEARER_STATE` reports.
16
17use std::fmt;
18use std::path::PathBuf;
19use std::sync::{Arc, Mutex};
20
21use tokio::sync::{mpsc, oneshot};
22use umsh_lan_bearer::socket::LanConfig;
23use umsh_ulcp_device::{QueuedClass, QueuedNotice};
24use umsh_ulcp_simdev::host::HostPlatform;
25use umsh_ulcp_simdev::lan::{self, HostEvent, StateLock};
26use umsh_ulcp_simdev::soft::SoftDevice;
27
28/// What the device reports itself as.
29const DEV_VERSION: &str = concat!("umsh-mobile-lan/", env!("CARGO_PKG_VERSION"));
30
31#[derive(Clone, Copy, Debug, PartialEq, Eq, uniffi::Error)]
32pub enum MobileVirtualRadioError {
33    /// The state directory could not be created or locked.
34    StateUnavailable,
35    /// Another virtual radio, in this process or another, is using the
36    /// state directory.
37    StateInUse,
38    /// The device's worker could not be started.
39    WorkerUnavailable,
40    /// The local socket could not be made.
41    SocketUnavailable,
42}
43
44impl fmt::Display for MobileVirtualRadioError {
45    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
46        f.write_str(match self {
47            Self::StateUnavailable => "the virtual radio's state directory is unavailable",
48            Self::StateInUse => "another virtual radio is using this state directory",
49            Self::WorkerUnavailable => "the virtual radio's worker could not start",
50            Self::SocketUnavailable => "the virtual radio's local socket could not be made",
51        })
52    }
53}
54
55impl std::error::Error for MobileVirtualRadioError {}
56
57/// What a queued frame is, as far as a notification is concerned.
58#[derive(Clone, Copy, Debug, PartialEq, Eq, uniffi::Enum)]
59pub enum MobileQueuedClass {
60    /// Addressed to the host.
61    Direct,
62    /// Multicast or broadcast.
63    Group,
64}
65
66/// A frame entered the inbound queue while no host was attached.
67#[derive(Clone, Copy, Debug, PartialEq, Eq, uniffi::Record)]
68pub struct MobileQueuedNoticeRecord {
69    pub class: MobileQueuedClass,
70    /// The source is muted: counts still stand, but the frame earns no
71    /// notification of its own.
72    pub muted: bool,
73    /// Frames in the queue after this one.
74    pub depth: u32,
75}
76
77impl From<QueuedNotice> for MobileQueuedNoticeRecord {
78    fn from(notice: QueuedNotice) -> Self {
79        Self {
80            class: match notice.class {
81                QueuedClass::Direct => MobileQueuedClass::Direct,
82                QueuedClass::Group => MobileQueuedClass::Group,
83            },
84            muted: notice.muted,
85            depth: u32::try_from(notice.depth).unwrap_or(u32::MAX),
86        }
87    }
88}
89
90/// What the device has for the platform, beyond a host's own replies.
91///
92/// Called on the device's worker thread; implementations hand the value
93/// to their own executor and return.
94#[uniffi::export(with_foreign)]
95pub trait MobileVirtualRadioListener: Send + Sync {
96    /// A frame was queued while no host was attached.
97    fn on_queued(&self, notice: MobileQueuedNoticeRecord);
98}
99
100/// Where one attached host's replies go.
101///
102/// Called on the device's worker thread. Each attach brings its own sink,
103/// and a sink hears nothing once its host detaches or is replaced, so it
104/// never carries the bytes of another session.
105#[uniffi::export(with_foreign)]
106pub trait MobileWireSink: Send + Sync {
107    /// HDLC-Lite bytes for the host.
108    fn on_wire(&self, bytes: Vec<u8>);
109}
110
111/// A running virtual radio. Dropping it, or calling [`Self::stop`], stops
112/// the device and releases its state directory.
113#[derive(uniffi::Object)]
114pub struct MobileVirtualRadio {
115    events: mpsc::UnboundedSender<HostEvent>,
116    /// Dropped to stop the device.
117    stop: Mutex<Option<oneshot::Sender<()>>>,
118}
119
120#[uniffi::export]
121impl MobileVirtualRadio {
122    /// Start a virtual radio keeping its saved configuration under
123    /// `state_dir`, for a host in this process.
124    ///
125    /// With `enable_phy`, the device comes up listening whatever it saved:
126    /// starting it is choosing the LAN. Without, its saved state decides,
127    /// so a host that turned the PHY off and saved keeps it off across
128    /// restarts.
129    #[uniffi::constructor]
130    pub fn start(
131        state_dir: String,
132        enable_phy: bool,
133        listener: Arc<dyn MobileVirtualRadioListener>,
134    ) -> Result<Arc<Self>, MobileVirtualRadioError> {
135        Self::launch(PathBuf::from(state_dir), None, enable_phy, listener)
136    }
137
138    /// Start a virtual radio as [`Self::start`] does, for hosts that
139    /// connect to the local socket at `socket_path`: connecting attaches,
140    /// closing detaches, and a new connection replaces the current one.
141    /// The socket is placed where only processes entitled to the radio
142    /// can reach it, and is removed when the radio stops.
143    #[uniffi::constructor]
144    pub fn serve(
145        state_dir: String,
146        socket_path: String,
147        enable_phy: bool,
148        listener: Arc<dyn MobileVirtualRadioListener>,
149    ) -> Result<Arc<Self>, MobileVirtualRadioError> {
150        Self::launch(
151            PathBuf::from(state_dir),
152            Some(PathBuf::from(socket_path)),
153            enable_phy,
154            listener,
155        )
156    }
157
158    /// A host in this process attached, taking replies at `sink`: the
159    /// device starts a session, announcing it with `CMD_SESSION_RESET`.
160    pub fn attach(&self, sink: Arc<dyn MobileWireSink>) {
161        let _ = self.events.send(HostEvent::Attach(Box::new(move |bytes| {
162            sink.on_wire(bytes)
163        })));
164    }
165
166    /// Bytes from the attached host, in HDLC-Lite, in any chunking.
167    pub fn receive(&self, bytes: Vec<u8>) {
168        let _ = self.events.send(HostEvent::Bytes(bytes));
169    }
170
171    /// The host went away: the device carries on detached.
172    pub fn detach(&self) {
173        let _ = self.events.send(HostEvent::Detach);
174    }
175
176    /// Stop the device. It cannot be started again; make another.
177    pub fn stop(&self) {
178        if let Ok(mut stop) = self.stop.lock() {
179            stop.take();
180        }
181    }
182}
183
184impl MobileVirtualRadio {
185    fn launch(
186        dir: PathBuf,
187        socket: Option<PathBuf>,
188        enable_phy: bool,
189        listener: Arc<dyn MobileVirtualRadioListener>,
190    ) -> Result<Arc<Self>, MobileVirtualRadioError> {
191        let platform =
192            HostPlatform::new(&dir).map_err(|_| MobileVirtualRadioError::StateUnavailable)?;
193        let lock = StateLock::try_acquire(&dir)
194            .map_err(|_| MobileVirtualRadioError::StateUnavailable)?
195            .ok_or(MobileVirtualRadioError::StateInUse)?;
196
197        let (events, host_events) = mpsc::unbounded_channel();
198        let socket_events = events.clone();
199        let (stop, stopped) = oneshot::channel::<()>();
200        let (ready_tx, ready_rx) = std::sync::mpsc::channel();
201        std::thread::Builder::new()
202            .name("umsh-virtual-radio".to_owned())
203            .spawn(move || {
204                let runtime = match tokio::runtime::Builder::new_current_thread()
205                    .enable_all()
206                    .build()
207                {
208                    Ok(runtime) => runtime,
209                    Err(_) => {
210                        let _ = ready_tx.send(Err(MobileVirtualRadioError::WorkerUnavailable));
211                        return;
212                    }
213                };
214                // The socket binds to this runtime's reactor, so it is
215                // made here rather than by the caller.
216                let local = match &socket {
217                    Some(path) => match runtime.block_on(async { lan::bind_local_socket(path) }) {
218                        Ok(local) => Some(local),
219                        Err(_) => {
220                            let _ = ready_tx.send(Err(MobileVirtualRadioError::SocketUnavailable));
221                            return;
222                        }
223                    },
224                    None => None,
225                };
226                let mut device = SoftDevice::new(lan::session_config(DEV_VERSION), platform);
227                if enable_phy && !device.session().settings().enabled {
228                    let _ = device.session_mut().toggle_radio(&mut |_| {});
229                }
230                let _ = ready_tx.send(Ok(()));
231                runtime.block_on(async move {
232                    // Boxed so the session lives on the heap, not this
233                    // thread's stack.
234                    let running = Box::pin(lan::run(
235                        device,
236                        LanConfig::lan(),
237                        host_events,
238                        move |notice| listener.on_queued(notice.into()),
239                    ));
240                    let serving = async move {
241                        if let Some(local) = local {
242                            lan::serve(local, socket_events).await;
243                        }
244                        std::future::pending::<()>().await
245                    };
246                    tokio::select! {
247                        () = running => {}
248                        () = serving => {}
249                        _ = stopped => {}
250                    }
251                });
252                if let Some(path) = socket {
253                    let _ = std::fs::remove_file(path);
254                }
255                drop(lock);
256            })
257            .map_err(|_| MobileVirtualRadioError::WorkerUnavailable)?;
258        ready_rx
259            .recv()
260            .map_err(|_| MobileVirtualRadioError::WorkerUnavailable)??;
261        Ok(Arc::new(Self {
262            events,
263            stop: Mutex::new(Some(stop)),
264        }))
265    }
266}
267
268#[cfg(test)]
269mod tests {
270    use std::sync::mpsc as std_mpsc;
271    use std::time::Duration;
272
273    use umsh_ulcp::{frame, hdlc, ids::prop};
274
275    use super::*;
276
277    struct Collect(Mutex<std_mpsc::Sender<Vec<u8>>>);
278
279    impl MobileVirtualRadioListener for Collect {
280        fn on_queued(&self, _: MobileQueuedNoticeRecord) {}
281    }
282
283    impl MobileWireSink for Collect {
284        fn on_wire(&self, bytes: Vec<u8>) {
285            let _ = self.0.lock().unwrap().send(bytes);
286        }
287    }
288
289    fn scratch(tag: &str) -> PathBuf {
290        let dir = std::env::temp_dir().join(format!(
291            "umsh-mobile-virtual-radio-{tag}-{}",
292            std::process::id()
293        ));
294        let _ = std::fs::remove_dir_all(&dir);
295        dir
296    }
297
298    /// A host's request is answered, and a second radio on the same state
299    /// is refused while the first runs.
300    #[test]
301    fn a_host_reaches_the_device() {
302        let dir = scratch("host");
303        let (wire_tx, wire) = std_mpsc::channel();
304        let listener = Arc::new(Collect(Mutex::new(wire_tx)));
305        let path = dir.to_string_lossy().into_owned();
306        let radio = MobileVirtualRadio::start(path.clone(), true, listener.clone()).unwrap();
307        assert_eq!(
308            MobileVirtualRadio::start(path.clone(), true, listener.clone()).err(),
309            Some(MobileVirtualRadioError::StateInUse)
310        );
311
312        radio.attach(listener.clone());
313        let mut request = [0u8; 16];
314        let len = frame::prop_get(&mut request, 1, prop::PHY_ENABLED).unwrap();
315        let mut encoded = vec![0; hdlc::max_encoded_len(len)];
316        let encoded_len = hdlc::encode_frame(&request[..len], &mut encoded).unwrap();
317        radio.receive(encoded[..encoded_len].to_vec());
318
319        let mut decoder = hdlc::Decoder::<512>::new();
320        let mut frames = Vec::new();
321        while frames.len() < 2 {
322            let bytes = wire
323                .recv_timeout(Duration::from_secs(2))
324                .expect("the device answers");
325            for byte in bytes {
326                if let Some(Ok(frame)) = decoder.push(byte) {
327                    frames.push(frame.to_vec());
328                }
329            }
330        }
331        // The answer to the GET: PHY on, as asked for at start.
332        assert_eq!(frames[1].last(), Some(&1));
333
334        radio.stop();
335        // The worker lets go of the state as it winds down.
336        let mut again = None;
337        for _ in 0..100 {
338            match MobileVirtualRadio::start(path.clone(), false, listener.clone()) {
339                Ok(radio) => {
340                    again = Some(radio);
341                    break;
342                }
343                Err(_) => std::thread::sleep(Duration::from_millis(10)),
344            }
345        }
346        assert!(again.is_some());
347        let _ = std::fs::remove_dir_all(&dir);
348    }
349
350    /// A serving radio greets a connection with a session reset, and
351    /// takes its socket with it when it stops.
352    #[test]
353    fn a_serving_radio_answers_on_its_socket() {
354        use std::io::Read;
355
356        let dir = scratch("serve");
357        let socket = dir.join("ulcp");
358        let (wire_tx, _) = std_mpsc::channel();
359        let listener = Arc::new(Collect(Mutex::new(wire_tx)));
360        let radio = MobileVirtualRadio::serve(
361            dir.to_string_lossy().into_owned(),
362            socket.to_string_lossy().into_owned(),
363            false,
364            listener,
365        )
366        .unwrap();
367        let mut stream = std::os::unix::net::UnixStream::connect(&socket).unwrap();
368        stream
369            .set_read_timeout(Some(Duration::from_secs(2)))
370            .unwrap();
371        let mut buf = [0u8; 64];
372        assert!(stream.read(&mut buf).unwrap() > 0);
373
374        radio.stop();
375        for _ in 0..100 {
376            if !socket.exists() {
377                break;
378            }
379            std::thread::sleep(Duration::from_millis(10));
380        }
381        assert!(!socket.exists());
382        let _ = std::fs::remove_dir_all(&dir);
383    }
384}
umsh_mobile_core/virtual_radio.rs

umsh_mobile_core/
virtual_radio.rs
