quantum_link/
worker.rs

1// SPDX-FileCopyrightText: 2025 Foundation Devices, Inc. <hello@foundation.xyz>
2// SPDX-License-Identifier: GPL-3.0-or-later
3
4use std::{future::Future, marker::PhantomData};
5
6use rkyv::bytecheck::CheckBytes;
7use server::{CheckedPermissions, MessageAllowed, XousDeserializer, XousValidator};
8use worker::{StreamWatch, WorkerHandle};
9
10use crate::{messages::SubscribeConnectionStatus, ConnectionStatus};
11
12/// reactive handle to latest QuantumLink status
13#[derive(Clone)]
14pub struct QlStatus<P> {
15    watch: StreamWatch<ConnectionStatus>,
16    worker: WorkerHandle,
17    _phantom: PhantomData<fn() -> P>,
18}
19
20impl<P> QlStatus<P>
21where
22    P: CheckedPermissions + 'static,
23{
24    pub fn new(worker: WorkerHandle) -> Self
25    where
26        P: MessageAllowed<SubscribeConnectionStatus> + 'static,
27    {
28        let sub = worker.subscribe_scalar::<P, _>(SubscribeConnectionStatus);
29        let initial = ConnectionStatus { bt_connected: false, ql_paired: false, live: false };
30        let watch = worker.watch_stream(sub, initial);
31        Self { watch, worker, _phantom: Default::default() }
32    }
33
34    /// wait until bluetooth is connected, device is paired, and connection is confirmed as live
35    pub async fn ready(&self) {
36        self.watch.wait_until(|status| status.bt_connected && status.ql_paired && status.live).await
37    }
38
39    /// wait until bluetooth is connected
40    pub async fn bt_ready(&self) { self.watch.wait_until(|status| status.bt_connected).await }
41
42    /// check if fully connected (BT + paired)
43    pub fn is_connected(&self) -> bool {
44        let status = self.watch.borrow();
45        status.bt_connected && status.ql_paired
46    }
47
48    // send a ql archive, after waiting for a connection
49    pub fn send_ql_archive<M>(&self, msg: M) -> impl Future<Output = M::Response>
50    where
51        P: server::MessageAllowed<M>,
52        M: server::BlockingArchive + Send + 'static,
53        M::Response: Send,
54    {
55        let this = self.clone();
56        async move {
57            this.ready().await;
58            this.worker.async_archive::<P, _>(msg).await
59        }
60    }
61
62    // retry publishing the message indefinitely
63    pub fn send_ql_archive_retry<M, T, E>(
64        &self,
65        msg: M,
66        mut error: impl FnMut(E) + Send + 'static,
67    ) -> impl Future<Output = T>
68    where
69        P: server::CheckedPermissions + server::MessageAllowed<M>,
70        M: server::BlockingArchive<Response = Result<T, E>> + Send + Clone + 'static,
71        M::Response: Send,
72        T: server::ArchiveCodec + Send + 'static,
73        <T as rkyv::Archive>::Archived:
74            rkyv::Deserialize<T, XousDeserializer> + for<'a> CheckBytes<XousValidator<'a>>,
75        E: server::ArchiveCodec + Send + 'static,
76        <E as rkyv::Archive>::Archived:
77            rkyv::Deserialize<E, XousDeserializer> + for<'a> CheckBytes<XousValidator<'a>>,
78    {
79        let this = self.clone();
80        async move {
81            loop {
82                this.ready().await;
83                match this.worker.async_archive::<P, _>(msg.clone()).await {
84                    Ok(value) => {
85                        return value;
86                    }
87                    Err(e) => {
88                        error(e);
89                    }
90                }
91            }
92        }
93    }
94
95    pub fn into_inner(self) -> StreamWatch<ConnectionStatus> { self.watch }
96}
97
98impl<P> std::ops::Deref for QlStatus<P> {
99    type Target = StreamWatch<ConnectionStatus>;
100
101    fn deref(&self) -> &Self::Target { &self.watch }
102}
103
104impl<P> std::ops::DerefMut for QlStatus<P> {
105    fn deref_mut(&mut self) -> &mut Self::Target { &mut self.watch }
106}
quantum_link/worker.rs

quantum_link/
worker.rs
