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