twinleaf/tio/
proxy.rs

1//! Proxy
2//!
3//! A proxy allows to multiplex access to a hardware port, by one or more
4//! `proxy::Port`s. These allow sending and receiving `proto::Packet`s
5//! to/from a single sensor, or an arbitrary device tree, and restrcting the
6//! type of traffic. Also, the proxy implements serial port rate negotiation
7//! for higher data rates.
8//!
9//! Note: the proxy runs in a dedicated thread.
10
11use super::port;
12use super::proto::{self, DeviceRoute, Packet, ProxyStatus};
13use super::proxy_core::{ProxyClient, ProxyCore};
14use super::util;
15use super::util::{TioRpcReplyable, TioRpcRequestable};
16
17use std::env;
18use std::thread;
19use std::time::Duration;
20
21use crossbeam::channel;
22
23/// Status event that ProxyCore sent back to an optional user specified channel
24#[derive(Debug)]
25pub enum Event {
26    SensorConnected,
27    SensorDisconnected,
28    SensorReconnected,
29    FailedToConnect,
30    FailedToReconnect,
31    Exiting,
32    ProtocolError(proto::Error),
33    FatalError(port::RecvError),
34    NewClient(u64),
35    RpcRemap((u64, u16), u16),
36    RpcRestore(u16, (u64, u16)),
37    RpcRestoreNotFound(u16),
38    RpcClientNotFound(u64),
39    RpcTimeout(u16),
40    RpcCancel(u16),
41    ClientSendFailed(u64),
42    ClientTerminated(u64),
43    RootDeviceRestarted,
44    AutoRateGaveUp,
45    AutoRateQueried(u32),
46    AutoRateRpcError(proto::RpcErrorCode),
47    AutoRateRpcInvalid,
48    AutoRateIncompatible(u32),
49    AutoRateCompatible(u32),
50    AutoRateWait,
51    AutoRateSet(u32),
52    SetRate(u32),
53    SetRateFailed,
54    NoData,
55}
56
57impl From<ProxyStatus> for super::proxy::Event {
58    fn from(status: ProxyStatus) -> Self {
59        match status {
60            ProxyStatus::SensorDisconnected => Event::SensorDisconnected,
61            ProxyStatus::SensorReconnected => Event::SensorReconnected,
62            ProxyStatus::FailedToReconnect => Event::FailedToReconnect,
63            ProxyStatus::FailedToConnect => Event::FailedToConnect,
64            ProxyStatus::Unknown(_) => Event::SensorDisconnected,
65        }
66    }
67}
68
69/// A port which communicates with a proxy via `crossbeam::channel`s
70pub struct Port {
71    tx: channel::Sender<Packet>,
72    rx: channel::Receiver<Packet>,
73    depth: usize,
74    scope: DeviceRoute,
75}
76
77#[derive(Debug, Clone, thiserror::Error)]
78pub enum SendError {
79    #[error("channel full")]
80    WouldBlock(Packet),
81    #[error("proxy disconnected")]
82    ProxyDisconnected(Packet),
83    #[error("route exceeds port scope")]
84    InvalidRoute(Packet),
85}
86
87#[derive(Debug, Clone, thiserror::Error)]
88pub enum RecvError {
89    #[error("no packet available")]
90    WouldBlock,
91    #[error("proxy disconnected")]
92    ProxyDisconnected,
93}
94
95#[derive(Debug, Clone, thiserror::Error)]
96pub enum RpcError {
97    #[error("failed to send RPC request: {0}")]
98    SendFailed(#[from] SendError),
99    #[error("device returned error: {0}")]
100    ExecError(proto::RpcErrorPayload),
101    #[error("failed to receive RPC reply: {0}")]
102    RecvFailed(#[from] RecvError),
103    #[error("RPC reply did not match expected type")]
104    TypeError,
105}
106
107impl Port {
108    /// Get the scope for this port
109    pub fn scope(&self) -> DeviceRoute {
110        self.scope.clone()
111    }
112
113    /// Sends a TIO packet to this port synchronously. This call will
114    /// block if the port is backed up.
115    pub fn send(&self, packet: Packet) -> Result<(), SendError> {
116        if packet.routing.len() > self.depth {
117            return Err(SendError::InvalidRoute(packet));
118        }
119        match self.tx.send(packet) {
120            Ok(()) => Ok(()),
121            Err(se) => Err(SendError::ProxyDisconnected(se.into_inner())),
122        }
123    }
124
125    /// Attempts to send a TIO packet to this port without blocking.
126    pub fn try_send(&self, packet: Packet) -> Result<(), SendError> {
127        if packet.routing.len() > self.depth {
128            return Err(SendError::InvalidRoute(packet));
129        }
130        match self.tx.try_send(packet) {
131            Ok(()) => Ok(()),
132            Err(crossbeam::channel::TrySendError::Full(pkt)) => Err(SendError::WouldBlock(pkt)),
133            Err(crossbeam::channel::TrySendError::Disconnected(pkt)) => {
134                Err(SendError::ProxyDisconnected(pkt))
135            }
136        }
137    }
138
139    /// `Select` the tx channel
140    pub fn select_send<'a>(&'a self, sel: &mut crossbeam::channel::Select<'a>) -> usize {
141        sel.send(&self.tx)
142    }
143
144    /// Waits for a packet to be available, and returns it.
145    pub fn recv(&self) -> Result<Packet, RecvError> {
146        match self.rx.recv() {
147            Ok(pkt) => Ok(pkt),
148            Err(crossbeam::channel::RecvError) => Err(RecvError::ProxyDisconnected),
149        }
150    }
151
152    /// Returns a packet if available, otherwise it doesn't stop.
153    pub fn try_recv(&self) -> Result<Packet, RecvError> {
154        match self.rx.try_recv() {
155            Ok(pkt) => Ok(pkt),
156            Err(crossbeam::channel::TryRecvError::Empty) => Err(RecvError::WouldBlock),
157            Err(crossbeam::channel::TryRecvError::Disconnected) => {
158                Err(RecvError::ProxyDisconnected)
159            }
160        }
161    }
162
163    /// `Select` the rx channel
164    pub fn select_recv<'a>(&'a self, sel: &mut crossbeam::channel::Select<'a>) -> usize {
165        sel.recv(&self.rx)
166    }
167
168    /// To use `crossbeam::channel::select!`.
169    pub fn receiver<'a>(&'a self) -> &'a crossbeam::channel::Receiver<Packet> {
170        &self.rx
171    }
172
173    /// Iterate over packets (until disconnect or break out).
174    pub fn iter(&self) -> crossbeam::channel::Iter<'_, Packet> {
175        self.rx.iter()
176    }
177
178    /// Iterate over packets (until disconnect, break out, or empty channel).
179    pub fn try_iter(&self) -> crossbeam::channel::TryIter<'_, Packet> {
180        self.rx.try_iter()
181    }
182
183    /// Generic any sized input/output RPC, blocking
184    pub fn raw_rpc(&self, name: &str, arg: &[u8]) -> Result<Vec<u8>, RpcError> {
185        if let Err(err) = self.send(util::PacketBuilder::make_rpc_request(
186            name,
187            arg,
188            0,
189            DeviceRoute::root(),
190        )) {
191            return Err(RpcError::SendFailed(err));
192        }
193        loop {
194            match self.recv() {
195                Ok(pkt) => match pkt.payload {
196                    proto::Payload::RpcReply(rep) => return Ok(rep.reply),
197                    proto::Payload::RpcError(err) => return Err(RpcError::ExecError(err)),
198                    _ => continue,
199                },
200                Err(err) => {
201                    return Err(RpcError::RecvFailed(err));
202                }
203            }
204        }
205    }
206
207    pub fn rpc<ReqT: TioRpcRequestable<ReqT>, RepT: TioRpcReplyable<RepT>>(
208        &self,
209        name: &str,
210        arg: ReqT,
211    ) -> Result<RepT, RpcError> {
212        let ret = self.raw_rpc(name, &arg.to_request())?;
213        if let Ok(val) = RepT::from_reply(&ret) {
214            Ok(val)
215        } else {
216            Err(RpcError::TypeError)
217        }
218    }
219
220    /// Action: rpc with no argument which returns nothing
221    pub fn action(&self, name: &str) -> Result<(), RpcError> {
222        self.rpc(name, ())
223    }
224
225    pub fn get<T: TioRpcReplyable<T>>(&self, name: &str) -> Result<T, RpcError> {
226        self.rpc(name, ())
227    }
228}
229
230#[derive(Debug, Clone, thiserror::Error)]
231pub enum PortError {
232    #[error("RPC timeout too short")]
233    RpcTimeoutTooShort,
234    #[error("RPC timeout too long")]
235    RpcTimeoutTooLong,
236    #[error("failed to set up new proxy client")]
237    FailedNewClientSetup,
238}
239
240/// Interface to a port proxy. Can create new ports.
241pub struct Interface {
242    new_client_queue: channel::Sender<ProxyClient>,
243    new_client_confirm: Option<channel::Receiver<Event>>,
244    client_rx_channel_size: usize,
245    client_tx_channel_size: usize,
246}
247
248impl Interface {
249    /// Create a new Interface, and a new ProxyCore running in a separate thread.
250    pub fn new_proxy(
251        url: &str,
252        reconnect_timeout: Option<Duration>,
253        status_queue: Option<channel::Sender<Event>>,
254    ) -> Interface {
255        let (client_sender, client_receiver) = channel::bounded::<ProxyClient>(5);
256        let (status_sender, status_receiver, only_clients) = {
257            if let Some(status_sender) = status_queue {
258                (status_sender, None, false)
259            } else {
260                let (s, r) = channel::bounded::<Event>(50);
261                (s, Some(r), true)
262            }
263        };
264        let url_string = url.to_string();
265        thread::spawn(move || {
266            #[cfg(target_os = "windows")]
267            let _priority = super::os::windows_helpers::ActivityGuard::latency_critical()
268                .map_err(|e| eprintln!("proxy core: failed to raise thread priority: {e}"))
269                .ok();
270
271            #[cfg(target_os = "macos")]
272            let _activity =
273                super::os::macos_helpers::ActivityGuard::latency_critical("Twinleaf proxy core");
274
275            let mut proxy = ProxyCore::new(
276                url_string,
277                reconnect_timeout,
278                client_receiver,
279                status_sender,
280                only_clients,
281            );
282            proxy.run();
283        });
284        Interface {
285            new_client_queue: client_sender,
286            new_client_confirm: status_receiver,
287            client_rx_channel_size: Self::get_client_rx_channel_size(),
288            client_tx_channel_size: Self::get_client_tx_channel_size(),
289        }
290    }
291
292    /// Create a new proxy which connects to a url with default parameters.
293    pub fn new(url: &str) -> Interface {
294        Self::new_proxy(url, None, None)
295    }
296
297    /// Create a new proxy which connects to a standalone tio-proxy process at
298    /// the default address.
299    pub fn default() -> Interface {
300        Self::new(util::default_proxy_url())
301    }
302
303    pub fn get_client_rx_channel_size() -> usize {
304        let min_size = port::DEFAULT_RX_CHANNEL_SIZE;
305        if let Ok(req) = env::var("TWINLEAF_PROXY_INTERFACE_RX_BUFSIZE") {
306            std::cmp::max(req.parse().unwrap_or(0), min_size)
307        } else {
308            min_size
309        }
310    }
311
312    pub fn get_client_tx_channel_size() -> usize {
313        let min_size = port::DEFAULT_TX_CHANNEL_SIZE;
314        if let Ok(req) = env::var("TWINLEAF_PROXY_INTERFACE_TX_BUFSIZE") {
315            std::cmp::max(req.parse().unwrap_or(0), min_size)
316        } else {
317            min_size
318        }
319    }
320
321    /// Create a new port
322    pub fn new_port(
323        &self,
324        rpc_timeout: Option<Duration>,
325        scope: DeviceRoute,
326        depth: usize,
327        forward_data: bool,
328        forward_nonrpc: bool,
329    ) -> Result<Port, PortError> {
330        let default_rpc_timeout = Duration::from_millis(3000);
331        let rpc_timeout = rpc_timeout.unwrap_or(default_rpc_timeout);
332        if rpc_timeout < Duration::from_millis(100) {
333            return Err(PortError::RpcTimeoutTooShort);
334        }
335        if rpc_timeout > Duration::from_secs(60) {
336            return Err(PortError::RpcTimeoutTooLong);
337        }
338
339        let (client_to_proxy_sender, proxy_from_client_receiver) =
340            channel::bounded::<Packet>(self.client_tx_channel_size);
341        let (proxy_to_client_sender, client_from_proxy_receiver) =
342            channel::bounded::<Packet>(self.client_rx_channel_size);
343        if let Err(_) = self.new_client_queue.send(ProxyClient::new(
344            proxy_to_client_sender,
345            proxy_from_client_receiver,
346            rpc_timeout,
347            scope.clone(),
348            depth,
349            forward_data,
350            forward_nonrpc,
351        )) {
352            return Err(PortError::FailedNewClientSetup);
353        }
354        if let Some(confirm) = &self.new_client_confirm {
355            if let Err(_) = confirm.recv() {
356                return Err(PortError::FailedNewClientSetup);
357            }
358        }
359        Ok(Port {
360            tx: client_to_proxy_sender,
361            rx: client_from_proxy_receiver,
362            depth: depth,
363            scope: scope,
364        })
365    }
366
367    /// New port with default parameters for a subtree, receiving all packets.
368    pub fn subtree_full(&self, subtree_root: DeviceRoute) -> Result<Port, PortError> {
369        self.new_port(None, subtree_root, usize::MAX, true, true)
370    }
371
372    /// New port with default parameters for a subtree, receiving only RPCs.
373    pub fn subtree_rpc(&self, subtree_root: DeviceRoute) -> Result<Port, PortError> {
374        self.new_port(None, subtree_root, usize::MAX, false, false)
375    }
376
377    /// New port with default parameters for a subtree, useful to probe connected devices.
378    pub fn subtree_probe(&self, subtree_root: DeviceRoute) -> Result<Port, PortError> {
379        self.new_port(None, subtree_root, usize::MAX, false, true)
380    }
381
382    /// New port for the full device tree, useful to probe connected devices.
383    pub fn tree_probe(&self) -> Result<Port, PortError> {
384        self.subtree_probe(DeviceRoute::root())
385    }
386
387    /// New port with default parameters for the full device tree, receiving all packets.
388    pub fn tree_full(&self) -> Result<Port, PortError> {
389        self.subtree_full(DeviceRoute::root())
390    }
391
392    /// New port with default parameters for the full device tree, receiving only RPCs.
393    pub fn tree_rpc(&self) -> Result<Port, PortError> {
394        self.subtree_rpc(DeviceRoute::root())
395    }
396
397    /// New port with default parameters for a specific device, receiving all packets.
398    pub fn device_full(&self, address: DeviceRoute) -> Result<Port, PortError> {
399        self.new_port(None, address, 0, true, true)
400    }
401
402    /// New port with default parameters for a specific device, receiving only RPCs.
403    pub fn device_rpc(&self, address: DeviceRoute) -> Result<Port, PortError> {
404        self.new_port(None, address, 0, false, false)
405    }
406
407    /// New port with default parameters for the root device, receiving all packets.
408    pub fn root_full(&self) -> Result<Port, PortError> {
409        self.device_full(DeviceRoute::root())
410    }
411
412    /// New port with default parameters for the root device, receiving only RPCs.
413    pub fn root_rpc(&self) -> Result<Port, PortError> {
414        self.device_rpc(DeviceRoute::root())
415    }
416}
twinleaf/tio/proxy.rs

twinleaf/tio/
proxy.rs
