kevy_client/
pipeline.rs

1//! Pipelined batches: queue N commands client-side, send
2//! them as one write, read N replies in order.
3//!
4//! Unlike [`Connection::multi`](crate::Connection::multi) this is
5//! **not** atomic — the server may interleave other clients' commands
6//! between the batch's — it only removes the per-command round-trip.
7//! Remote-only: the embedded backend has no argv dispatcher (same
8//! rationale as MULTI) and its calls are in-process already; it
9//! answers `Unsupported`.
10//!
11//! ```no_run
12//! use kevy_client::Connection;
13//!
14//! let mut conn = Connection::connect("kevy://localhost:6379")?;
15//! let replies = conn.pipeline(|p| {
16//!     p.cmd(&[b"SET", b"a", b"1"]).cmd(&[b"INCR", b"n"]).cmd(&[b"GET", b"a"]);
17//! })?;
18//! assert_eq!(replies.len(), 3);
19//! # Ok::<(), kevy_client::KevyError>(())
20//! ```
21
22use crate::{KevyError, KevyResult};
23
24use kevy_resp::{Reply, encode_command_borrowed};
25
26use crate::Connection;
27
28/// Client-side command buffer for [`Connection::pipeline`]. Each
29/// [`Self::cmd`] appends one RESP-encoded command; the whole buffer is
30/// flushed in a single write.
31#[derive(Debug, Default)]
32pub struct PipelineBuf {
33    buf: Vec<u8>,
34    count: usize,
35    /// Set when a `cmd` call was malformed (empty argv) — surfaced as
36    /// `InvalidInput` at send time so the builder closure can stay
37    /// infallible-chainable.
38    poisoned: bool,
39}
40
41impl PipelineBuf {
42    /// Queue one command — verb + args as raw byte slices, e.g.
43    /// `p.cmd(&[b"SET", key, value])`. Chainable. An empty `parts`
44    /// poisons the buffer and [`Connection::pipeline`] reports it.
45    pub fn cmd(&mut self, parts: &[&[u8]]) -> &mut Self {
46        if parts.is_empty() {
47            self.poisoned = true;
48            return self;
49        }
50        encode_command_borrowed(&mut self.buf, parts);
51        self.count += 1;
52        self
53    }
54
55    /// Number of queued commands.
56    pub fn len(&self) -> usize {
57        self.count
58    }
59
60    /// Whether nothing has been queued.
61    pub fn is_empty(&self) -> bool {
62        self.count == 0
63    }
64}
65
66impl Connection {
67    /// Run a pipelined batch: `build` queues commands into a
68    /// [`PipelineBuf`], then the batch is sent as one write and the
69    /// per-command replies are returned **in queue order**.
70    ///
71    /// Error replies to individual commands come back as
72    /// [`Reply::Error`] entries (they don't abort the batch) — check
73    /// each reply, unlike the typed single-command wraps which map
74    /// server errors to `io::Error`. An empty batch returns `Ok(vec![])`
75    /// without touching the wire.
76    pub fn pipeline(
77        &mut self,
78        build: impl FnOnce(&mut PipelineBuf),
79    ) -> KevyResult<Vec<Reply>> {
80        let client = self.remote("pipeline")?;
81        let mut p = PipelineBuf::default();
82        build(&mut p);
83        if p.poisoned {
84            return Err(KevyError::InvalidInput("pipeline: a queued command had an empty argv".into()));
85        }
86        if p.count == 0 {
87            return Ok(Vec::new());
88        }
89        Ok(client.pipeline_raw(&p.buf, p.count)?)
90    }
91}
92
93#[cfg(test)]
94mod tests {
95    use super::*;
96
97    #[test]
98    fn embedded_pipeline_unsupported() {
99        let mut c = Connection::connect("mem://").unwrap();
100        let err = c.pipeline(|p| {
101            p.cmd(&[b"PING"]);
102        });
103        assert!(matches!(err.unwrap_err(), KevyError::Unsupported(_)));
104    }
105
106    #[test]
107    fn pipeline_buf_counts_and_encodes() {
108        let mut p = PipelineBuf::default();
109        assert!(p.is_empty());
110        p.cmd(&[b"SET", b"a", b"1"]).cmd(&[b"GET", b"a"]);
111        assert_eq!(p.len(), 2);
112        assert!(p.buf.starts_with(b"*3\r\n$3\r\nSET\r\n"));
113    }
114}
kevy_client/pipeline.rs

kevy_client/
pipeline.rs
