thruput_rpc_api/
sender.rs

1//! Trait for making outgoing RPC calls.
2
3use crate::{CallBuf, ProgVers, RpcError, RpcResponse};
4use async_trait::async_trait;
5
6/// Trait for making outgoing RPC calls.
7///
8/// `thruput_rpc_driver` implements this for real TCP connections
9/// (`ConnectionHandle`, `ClientConnection`) and for its pooled connections
10/// (`FixedSerialPool`, `DynamicPool`, `CpuSetPool`). Inside an
11/// [`RpcHandler`](crate::RpcHandler) dispatch, the per-connection sender is
12/// available via `thruput_rpc_driver::peer()`.
13///
14/// Generated protocol clients hold an `Arc<dyn RpcSender>` and are agnostic
15/// to which implementation backs it. Implement on a mock struct for unit
16/// testing without TCP.
17#[async_trait]
18pub trait RpcSender: Send + Sync {
19    /// A buffer to encode the next call into.
20    ///
21    /// Senders that recycle record buffers hand back a spent one, so a
22    /// steady stream of calls allocates nothing. A sender with nothing to
23    /// recycle returns [`CallBuf::new`]. Encode the arguments into the
24    /// buffer and pass it to [`send_call`](Self::send_call).
25    fn call_buf(&self) -> CallBuf;
26
27    /// Send an RPC call and return the response.
28    ///
29    /// The call's arguments arrive encoded in a [`CallBuf`], which the
30    /// driver frames in place and hands to the socket without copying. The
31    /// returned [`RpcResponse`] derefs to `&[u8]` for zero-copy
32    /// deserialization, or can be copied to an owned `Vec<u8>` via
33    /// `.to_vec()`.
34    async fn send_call(
35        &self,
36        progvers: ProgVers,
37        proc: u32,
38        call: CallBuf,
39    ) -> Result<RpcResponse, RpcError>;
40
41    /// Send a call and run `decode` on the reply's payload, which an
42    /// implementation may pass straight from its receive buffer.
43    ///
44    /// `decode` runs at most once, only for a successful reply. The
45    /// implementation stops using it before this future returns or finishes
46    /// being dropped. The default awaits [`send_call`](Self::send_call) and
47    /// decodes the owned response in the caller's task; transports override
48    /// it to skip that copy, running `decode` on their reader instead.
49    ///
50    /// # Safety
51    ///
52    /// Do not leak this future (`mem::forget`, a leaked box) and then free or
53    /// reuse what `decode` borrows. An implementation may call `decode` from
54    /// another thread until this future completes or its drop returns, and
55    /// leaking skips the drop that stops it. Dropping the future is fine.
56    ///
57    /// `decode` may run on the connection's reader, so it must not block, and
58    /// above all must not wait for another call on the same connection,
59    /// whose reply that reader would have to deliver.
60    ///
61    /// The rest is enforced by the signature: `decode` and what it captures
62    /// stay alive and unaliased while borrowed, and the `for<'reply>` bound
63    /// stops it from keeping the slice.
64    async unsafe fn send_call_decoded(
65        &self,
66        progvers: ProgVers,
67        proc: u32,
68        call: CallBuf,
69        decode: &mut (dyn for<'reply> FnMut(&'reply [u8]) -> Result<(), RpcError> + Send),
70    ) -> Result<(), RpcError> {
71        let response = self.send_call(progvers, proc, call).await?;
72        decode(&response)
73    }
74
75    /// Send an RPC call without waiting for the response.
76    ///
77    /// Default implementation calls [`send_call`](Self::send_call) and
78    /// discards the response. Implementations may override this to return
79    /// after queueing the request, in which case success does not confirm
80    /// socket delivery or peer-side dispatch.
81    async fn send_fire_and_forget(
82        &self,
83        progvers: ProgVers,
84        proc: u32,
85        call: CallBuf,
86    ) -> Result<(), RpcError> {
87        self.send_call(progvers, proc, call).await.map(|_| ())
88    }
89}
thruput_rpc_api/sender.rs

thruput_rpc_api/
sender.rs
