kevy_client/
pipeline.rs

1//! Pipelined batches (v1.14.0): 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::open("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::<(), std::io::Error>(())
20//! ```
21
22use std::io;
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    ) -> io::Result<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(io::Error::new(
85                io::ErrorKind::InvalidInput,
86                "pipeline: a queued command had an empty argv",
87            ));
88        }
89        if p.count == 0 {
90            return Ok(Vec::new());
91        }
92        client.pipeline_raw(&p.buf, p.count)
93    }
94}
95
96#[cfg(test)]
97mod tests {
98    use super::*;
99
100    #[test]
101    fn embedded_pipeline_unsupported() {
102        let mut c = Connection::open("mem://").unwrap();
103        let err = c.pipeline(|p| {
104            p.cmd(&[b"PING"]);
105        });
106        assert_eq!(err.unwrap_err().kind(), io::ErrorKind::Unsupported);
107    }
108
109    #[test]
110    fn pipeline_buf_counts_and_encodes() {
111        let mut p = PipelineBuf::default();
112        assert!(p.is_empty());
113        p.cmd(&[b"SET", b"a", b"1"]).cmd(&[b"GET", b"a"]);
114        assert_eq!(p.len(), 2);
115        assert!(p.buf.starts_with(b"*3\r\n$3\r\nSET\r\n"));
116    }
117}
kevy_client/pipeline.rs

kevy_client/
pipeline.rs
