darkbio_wire/protocol/requester.rs
1// wire-rs: encrypted protocol between Ark and host
2// Copyright 2026 Dark Bio AG. All rights reserved.
3
4//! Sending requests through a shared handle to a session.
5
6use super::session::SessionInner;
7use super::{Error, Message, Promise};
8use std::sync::Weak;
9use std::time::Instant;
10
11/// Clonable handle for sending requests through the session that created it.
12/// Does not keep the session open or follow a replacement session. Dropping a
13/// requester does not close the session or cancel operations it already submitted.
14#[derive(Clone)]
15pub struct Requester {
16 /// Session that created this requester, even after a replacement connects.
17 session: Weak<SessionInner>,
18}
19
20impl Requester {
21 /// Queues a request and returns its promise without waiting for the writer or
22 /// a reply. A closed session returns an error immediately; errors after queueing
23 /// are returned through the promise. The outgoing queue has no capacity limit.
24 /// A message invalid for this session's direction fails the promise with
25 /// [`Error::WrongDirection`].
26 ///
27 /// The deadline covers time in the queue, sending and accepting the response.
28 /// Waiting on the promise does not start or refresh it. Dropping the promise
29 /// does not cancel the request. The peer may keep working after a timeout.
30 /// Decoding the response in `wait()` is outside this deadline.
31 /// The expected response type is selected when waiting on [`Promise<Message>`].
32 pub fn request(
33 &self,
34 request: impl Into<Message>,
35 deadline: Instant,
36 ) -> Result<Promise<Message>, Error> {
37 self.session
38 .upgrade()
39 .ok_or(Error::Closed)?
40 .request(request.into(), deadline)
41 }
42
43 /// Creates a requester from a weak reference to its session.
44 pub(super) fn new(session: Weak<SessionInner>) -> Self {
45 Self { session }
46 }
47}
48
49/// Checks requester sharing and compiles pipelined request submission.
50#[cfg(test)]
51#[cfg_attr(coverage_nightly, coverage(off))]
52mod tests {
53 use crate::protocol::{
54 DeviceInfoRequest, DeviceInfoResponse, Error, Message, Promise, Requester, Session,
55 };
56 use std::time::Instant;
57
58 /// Compiles sending several requests before waiting, dropping promises, and
59 /// choosing the expected response type at `wait()`.
60 #[allow(dead_code)]
61 fn pipeline(session: &Session, deadline: Instant) -> Result<(), Error> {
62 let requester: Requester = session.requester();
63 let first: Promise<Message> = requester.request(DeviceInfoRequest {}, deadline)?;
64 let second = requester.request(DeviceInfoRequest {}, deadline)?;
65
66 // A promise can be dropped without selecting a response type.
67 drop(requester.request(DeviceInfoRequest {}, deadline)?);
68
69 // Caller-selected typing, by annotation or by explicit generic argument.
70 let _: DeviceInfoResponse = second.wait()?;
71 let _ = first.wait::<DeviceInfoResponse>()?;
72
73 // Callers may also request the message enum to match it themselves.
74 let _: Message = requester.request(DeviceInfoRequest {}, deadline)?.wait()?;
75 Ok(())
76 }
77
78 /// Checks that `Requester` implements `Clone`, `Send`, and `Sync`.
79 #[test]
80 fn test_thread_capabilities() {
81 /// Requires a handle to be clonable and usable by multiple threads.
82 fn shared<T: Clone + Send + Sync + 'static>() {}
83 shared::<Requester>();
84 }
85}