rpc_api/
chunk.rs

1//! The wire-level vocabulary of a chunked RPC: where a record sits in its
2//! run, and how a chunked record comes apart into key, position and part.
3//! Reassembling a run is the driver's job (`rpc_driver::ChunkRunAssembler`);
4//! this crate holds only what a protocol crate needs to describe its chunked
5//! records, so generated bindings keep their single dependency on it.
6
7/// Where one record sits in the run of records that together carry a
8/// message too large for one RPC record: its index, from 0, and the run's
9/// length. Every record of a run carries the same `count`, each index in
10/// `[0, count)` is sent exactly once, and a run always has at least one
11/// record, so an empty message is a run of one empty record. The sender may
12/// issue a run's calls concurrently and over several connections, so its
13/// records can arrive in any order, and the runs of several messages can be
14/// in flight at once; the driver's `ChunkRunAssembler` puts them back
15/// together.
16#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
17pub struct ChunkPosition {
18    /// The record's index in its run, from 0.
19    pub index: u32,
20    /// The number of records in the run; at least 1.
21    pub count: u32,
22}
23
24impl ChunkPosition {
25    /// The position of record `index` in a run of `count` records.
26    pub const fn new(index: u32, count: u32) -> Self {
27        Self { index, count }
28    }
29}
30
31/// A wire record that is one chunk of a run: it names the message it is
32/// part of (its key, such as a batch number), its position in the run, and
33/// carries its part of the message's items.
34///
35/// Protocol crates implement this for each chunked record they define, so a
36/// handler passes the record it was called with straight to the driver's
37/// `ChunkRunAssembler::push_record`. It lives here so a protocol crate needs
38/// no transport dependency to implement it.
39///
40/// ```
41/// use rpc_api::{ChunkPosition, ChunkRecord};
42///
43/// // The shape a chunked wire record takes: the message it belongs to, its
44/// // position in the run, and its slice of the items.
45/// struct UpdateChunk {
46///     batch: i64,
47///     chunk_index: u32,
48///     chunk_count: u32,
49///     updates: Vec<i64>,
50/// }
51///
52/// impl ChunkRecord for UpdateChunk {
53///     type Key = i64;
54///     type Part = Vec<i64>;
55///
56///     fn into_chunk(self) -> (i64, ChunkPosition, Vec<i64>) {
57///         let position = ChunkPosition::new(self.chunk_index, self.chunk_count);
58///         (self.batch, position, self.updates)
59///     }
60/// }
61///
62/// let chunk = UpdateChunk { batch: 7, chunk_index: 1, chunk_count: 2, updates: vec![3] };
63/// assert_eq!(chunk.into_chunk(), (7, ChunkPosition::new(1, 2), vec![3]));
64/// ```
65pub trait ChunkRecord {
66    /// Identifies the message the run carries.
67    type Key;
68    /// One chunk's payload.
69    type Part;
70
71    /// Takes the record apart into its key, its position and its part.
72    fn into_chunk(self) -> (Self::Key, ChunkPosition, Self::Part);
73}
rpc_api/chunk.rs

rpc_api/
chunk.rs
