cranpose_services/
wearable_framed.rs

1//! A [`WearableLink`] over a platform link that carries only whole messages,
2//! such as WatchConnectivity between an iPhone app and its watch app.
3//!
4//! Each message the platform carries is one frame, and its first byte says
5//! what it holds:
6//!
7//! - an application message: the path's length (2 bytes, little-endian), the
8//!   path, then the data;
9//! - the start of a stream: the stream's number (4 bytes, little-endian), then
10//!   the path;
11//! - the next bytes of a stream: the stream's number, then the bytes;
12//! - the end of a stream: the stream's number.
13//!
14//! A stream writer gathers small writes into frames of up to [`CHUNK_BYTES`],
15//! so a microphone that writes every few milliseconds sends a few messages a
16//! second.
17
18use std::{
19    collections::HashMap,
20    io::{self, Read, Write},
21    sync::{
22        Arc,
23        atomic::{AtomicU32, Ordering},
24        mpsc::{Receiver, SyncSender, sync_channel},
25    },
26};
27
28use parking_lot::Mutex;
29
30use crate::wearable::{
31    WearableError, WearableEvent, WearableLink, WearablePeer, WearableStream,
32    publish_wearable_event,
33};
34
35/// The most bytes of a stream one frame carries. At 16 kHz, 16-bit mono
36/// audio this is half a second: two messages a second. WatchConnectivity
37/// between simulators fell behind eight a second, and each message wakes the
38/// watch's radio.
39pub const CHUNK_BYTES: usize = 16 * 1024;
40
41/// How many frames of a stream may wait for its reader before delivery of
42/// the next one waits too.
43const PENDING_CHUNKS: usize = 256;
44
45const MESSAGE: u8 = 0;
46const OPEN: u8 = 1;
47const CHUNK: u8 = 2;
48const END: u8 = 3;
49
50/// The kind byte and the stream's number.
51const STREAM_HEADER: usize = 5;
52
53/// What the platform's message link does for a [`FramedWearableLink`].
54pub trait WearableTransport: Send + Sync + 'static {
55    /// The connected devices.
56    fn peers(&self) -> Result<Vec<WearablePeer>, WearableError>;
57
58    /// Delivers one frame to `peer`; frames arrive in the order they were
59    /// sent. With `confirm` the call returns once the application on `peer`
60    /// has the frame. Without it the call returns once the platform took the
61    /// frame, and reports the failure of an earlier frame sent this way.
62    fn send(&self, peer: &str, frame: &[u8], confirm: bool) -> Result<(), WearableError>;
63
64    /// Opens `url` on `peer`'s screen.
65    fn open_url(&self, peer: &str, url: &str) -> Result<(), WearableError>;
66}
67
68/// Application messages and byte streams over a [`WearableTransport`].
69///
70/// The platform backend installs it with
71/// [`set_platform_wearable_link`](crate::set_platform_wearable_link) and hands
72/// it every frame the platform receives through [`Self::receive`].
73pub struct FramedWearableLink<T> {
74    transport: Arc<T>,
75    next_stream: AtomicU32,
76    incoming: Mutex<HashMap<String, Streams>>,
77}
78
79/// The open streams from one device, by their number.
80type Streams = HashMap<u32, SyncSender<Vec<u8>>>;
81
82impl<T: WearableTransport> FramedWearableLink<T> {
83    /// A link that sends its frames through `transport`.
84    pub fn new(transport: T) -> Self {
85        Self {
86            transport: Arc::new(transport),
87            next_stream: AtomicU32::new(0),
88            incoming: Mutex::new(HashMap::new()),
89        }
90    }
91
92    /// Handles one frame `peer` sent. A frame for a stream waits while that
93    /// stream's reader is far behind.
94    pub fn receive(&self, peer: &str, frame: &[u8]) {
95        let Some((&kind, body)) = frame.split_first() else {
96            log::warn!("wearable: an empty frame from {peer}");
97            return;
98        };
99        if kind == MESSAGE {
100            match split_path(body) {
101                Some((path, data)) => publish_wearable_event(WearableEvent::Message {
102                    peer: peer.to_owned(),
103                    path: path.to_owned(),
104                    data: data.to_vec(),
105                }),
106                None => log::warn!("wearable: a message frame without its path from {peer}"),
107            }
108            return;
109        }
110        let Some((id, body)) = split_stream(body) else {
111            log::warn!("wearable: a stream frame without its number from {peer}");
112            return;
113        };
114        match kind {
115            OPEN => self.open_incoming(peer, id, body),
116            CHUNK => {
117                // A clone, so the lock is free while the reader catches up.
118                let sender = self
119                    .incoming
120                    .lock()
121                    .get(peer)
122                    .and_then(|streams| streams.get(&id))
123                    .cloned();
124                if let Some(sender) = sender
125                    && sender.send(body.to_vec()).is_err()
126                {
127                    self.end_incoming(peer, id);
128                }
129            }
130            END => self.end_incoming(peer, id),
131            _ => log::warn!("wearable: a frame of unknown kind {kind} from {peer}"),
132        }
133    }
134
135    /// Ends every stream from `peer`, whose readers then reach the end of
136    /// their data. The backend calls it when the device goes out of reach.
137    pub fn disconnect(&self, peer: &str) {
138        self.incoming.lock().remove(peer);
139    }
140
141    fn open_incoming(&self, peer: &str, id: u32, path: &[u8]) {
142        let Ok(path) = std::str::from_utf8(path) else {
143            log::warn!("wearable: a stream from {peer} with a path that is not UTF-8");
144            return;
145        };
146        let (sender, chunks) = sync_channel(PENDING_CHUNKS);
147        // A number the device used before belongs to a stream it lost, which
148        // ends here.
149        self.incoming
150            .lock()
151            .entry(peer.to_owned())
152            .or_default()
153            .insert(id, sender);
154        publish_wearable_event(WearableEvent::Stream(WearableStream::new(
155            peer.to_owned(),
156            path.to_owned(),
157            Box::new(ChunkReader {
158                chunks,
159                chunk: Vec::new(),
160                read: 0,
161            }),
162        )));
163    }
164
165    fn end_incoming(&self, peer: &str, id: u32) {
166        if let Some(streams) = self.incoming.lock().get_mut(peer) {
167            streams.remove(&id);
168        }
169    }
170}
171
172impl<T: WearableTransport> WearableLink for FramedWearableLink<T> {
173    fn peers(&self) -> Result<Vec<WearablePeer>, WearableError> {
174        self.transport.peers()
175    }
176
177    fn send_message(&self, peer: &str, path: &str, data: &[u8]) -> Result<(), WearableError> {
178        let length = u16::try_from(path.len())
179            .map_err(|_| WearableError::Failed(format!("the path {path} is too long")))?;
180        let mut frame = Vec::with_capacity(3 + path.len() + data.len());
181        frame.push(MESSAGE);
182        frame.extend_from_slice(&length.to_le_bytes());
183        frame.extend_from_slice(path.as_bytes());
184        frame.extend_from_slice(data);
185        self.transport.send(peer, &frame, true)
186    }
187
188    fn open_stream(&self, peer: &str, path: &str) -> Result<Box<dyn Write + Send>, WearableError> {
189        let id = self.next_stream.fetch_add(1, Ordering::Relaxed);
190        let mut frame = Vec::with_capacity(STREAM_HEADER + CHUNK_BYTES.max(path.len()));
191        frame.push(OPEN);
192        frame.extend_from_slice(&id.to_le_bytes());
193        frame.extend_from_slice(path.as_bytes());
194        self.transport.send(peer, &frame, true)?;
195        frame[0] = CHUNK;
196        frame.truncate(STREAM_HEADER);
197        Ok(Box::new(ChunkWriter {
198            transport: Arc::clone(&self.transport),
199            peer: peer.to_owned(),
200            frame,
201        }))
202    }
203
204    fn open_url(&self, peer: &str, url: &str) -> Result<(), WearableError> {
205        self.transport.open_url(peer, url)
206    }
207}
208
209fn split_path(body: &[u8]) -> Option<(&str, &[u8])> {
210    let (length, rest) = body.split_first_chunk::<2>()?;
211    let (path, data) = rest.split_at_checked(usize::from(u16::from_le_bytes(*length)))?;
212    Some((std::str::from_utf8(path).ok()?, data))
213}
214
215fn split_stream(body: &[u8]) -> Option<(u32, &[u8])> {
216    let (id, rest) = body.split_first_chunk::<4>()?;
217    Some((u32::from_le_bytes(*id), rest))
218}
219
220/// The sending half of a stream: one frame it fills and sends when full, on
221/// [`Write::flush`], and when dropped.
222struct ChunkWriter<T: WearableTransport> {
223    transport: Arc<T>,
224    peer: String,
225    frame: Vec<u8>,
226}
227
228impl<T: WearableTransport> ChunkWriter<T> {
229    fn send_chunk(&mut self) -> io::Result<()> {
230        if self.frame.len() == STREAM_HEADER {
231            return Ok(());
232        }
233        let sent = self.transport.send(&self.peer, &self.frame, false);
234        self.frame.truncate(STREAM_HEADER);
235        sent.map_err(io::Error::other)
236    }
237}
238
239impl<T: WearableTransport> Write for ChunkWriter<T> {
240    fn write(&mut self, data: &[u8]) -> io::Result<usize> {
241        let room = STREAM_HEADER + CHUNK_BYTES - self.frame.len();
242        let taken = data.len().min(room);
243        self.frame.extend_from_slice(&data[..taken]);
244        if self.frame.len() == STREAM_HEADER + CHUNK_BYTES {
245            self.send_chunk()?;
246        }
247        Ok(taken)
248    }
249
250    fn flush(&mut self) -> io::Result<()> {
251        self.send_chunk()
252    }
253}
254
255impl<T: WearableTransport> Drop for ChunkWriter<T> {
256    fn drop(&mut self) {
257        if let Err(error) = self.send_chunk() {
258            log::warn!(
259                "wearable: the last bytes of a stream to {}: {error}",
260                self.peer
261            );
262        }
263        self.frame[0] = END;
264        if let Err(error) = self.transport.send(&self.peer, &self.frame, false) {
265            log::warn!("wearable: the end of a stream to {}: {error}", self.peer);
266        }
267    }
268}
269
270/// The receiving half of a stream.
271struct ChunkReader {
272    chunks: Receiver<Vec<u8>>,
273    chunk: Vec<u8>,
274    read: usize,
275}
276
277impl Read for ChunkReader {
278    fn read(&mut self, out: &mut [u8]) -> io::Result<usize> {
279        if out.is_empty() {
280            return Ok(0);
281        }
282        while self.read == self.chunk.len() {
283            match self.chunks.recv() {
284                Ok(chunk) => {
285                    self.chunk = chunk;
286                    self.read = 0;
287                }
288                // The sender ended the stream, or the link lost the device.
289                Err(_) => return Ok(0),
290            }
291        }
292        let count = out.len().min(self.chunk.len() - self.read);
293        out[..count].copy_from_slice(&self.chunk[self.read..self.read + count]);
294        self.read += count;
295        Ok(count)
296    }
297}
298
299#[cfg(test)]
300#[path = "tests/wearable_framed_tests.rs"]
301mod tests;
cranpose_services/wearable_framed.rs

cranpose_services/
wearable_framed.rs
