glide/
pipeline.rs

1// Copyright Valkey GLIDE Project Contributors - SPDX Identifier: Apache-2.0
2//! GLIDE's owned pipeline / transaction builder.
3//!
4//! A [`Pipeline`] batches commands into a single round-trip.
5
6use crate::ValkeyResult;
7use crate::cmd::Cmd;
8use crate::error::GlideError;
9use crate::pipeline_options::PipelineOptions;
10use crate::value::ValkeyValue;
11use crate::write::ToValkeyArgs;
12use crate::write::ValkeyWrite;
13use glide_core::client::Client as CoreClient;
14use redis::PipelineRetryStrategy;
15use redis::cluster_routing::RoutingInfo;
16
17/// Create an empty pipeline.
18///
19/// Mirrors redis-rs's `pipe`.
20pub fn pipe() -> Pipeline {
21    Pipeline::new()
22}
23
24/// A pipeline that batches commands into a single round-trip.
25///
26/// ```rust,no_run
27/// use glide::{PipelineExt, pipe};
28/// # async fn demo(client: &glide::GlideClient) -> glide::ValkeyResult<()> {
29/// let (a, b): (i64, i64) = pipe()
30///     .atomic()
31///     .incr("c", 1)
32///     .incr("c", 1)
33///     .query_async(client)
34///     .await?;
35/// # let _ = (a, b); Ok(()) }
36/// ```
37///
38/// Mirrors redis-rs's `Pipeline`.
39#[derive(Clone, Debug, Default)]
40pub struct Pipeline {
41    inner: redis::Pipeline,
42}
43
44impl Pipeline {
45    /// Create an empty pipeline.
46    pub fn new() -> Self {
47        Pipeline {
48            inner: redis::Pipeline::new(),
49        }
50    }
51
52    /// Enable atomic mode: the whole pipeline
53    /// runs as a `MULTI`/`EXEC` transaction.
54    pub fn atomic(&mut self) -> &mut Self {
55        self.inner.atomic();
56        self
57    }
58
59    /// Start a new command with the given keyword.
60    pub fn cmd(&mut self, name: &str) -> &mut Self {
61        self.inner.cmd(name);
62        self
63    }
64
65    /// Append argument(s) to the last started command.
66    pub fn arg<A: ToValkeyArgs>(&mut self, arg: A) -> &mut Self {
67        arg.write_valkey_args(self);
68        self
69    }
70
71    /// Ignore the last command's reply: it is still checked
72    /// for errors, but dropped from the typed response.
73    pub fn ignore(&mut self) -> &mut Self {
74        self.inner.ignore();
75        self
76    }
77
78    /// Push a fully-built [`Cmd`].
79    pub(crate) fn add_command(&mut self, cmd: Cmd) -> &mut Self {
80        self.inner.add_command(cmd.into_redis());
81        self
82    }
83
84    /// Borrow the underlying `redis::Pipeline` (feeds glide-core dispatch, and
85    /// exposes `ignored_commands()` for the typed-decode reply filtering).
86    pub(crate) fn as_redis(&self) -> &redis::Pipeline {
87        &self.inner
88    }
89}
90
91/// Writes arguments into the last started command.
92impl ValkeyWrite for Pipeline {
93    fn write_arg(&mut self, arg: &[u8]) {
94        self.inner.arg(arg);
95    }
96
97    fn write_arg_fmt(&mut self, arg: impl std::fmt::Display) {
98        self.inner.arg(arg.to_string().into_bytes());
99    }
100}
101
102/// Dispatches a [`Pipeline`] and returns the reply as a [`ValkeyValue`].
103pub(crate) async fn dispatch_pipeline(
104    core: &CoreClient,
105    pipeline: &Pipeline,
106    routing: Option<RoutingInfo>,
107    raise_on_error: bool,
108    options: &PipelineOptions,
109) -> ValkeyResult<ValkeyValue> {
110    let redis_pipeline = pipeline.as_redis();
111
112    // Return empty array directly for an empty pipeline.
113    if redis_pipeline.is_empty() {
114        return Ok(ValkeyValue::Array(Vec::new()));
115    }
116
117    let timeout = options.timeout_millis();
118    let mut client = core.clone();
119
120    let reply = if redis_pipeline.is_atomic() {
121        client
122            .send_transaction(redis_pipeline, routing, timeout, raise_on_error)
123            .await
124    } else {
125        let retry_strategy = PipelineRetryStrategy {
126            retry_server_error: options.retry_server_error,
127            retry_connection_error: options.retry_connection_error,
128        };
129        client
130            .send_pipeline(
131                redis_pipeline,
132                routing,
133                raise_on_error,
134                timeout,
135                retry_strategy,
136            )
137            .await
138    };
139
140    ValkeyValue::from_redis(reply.map_err(GlideError::from_redis_error)?)
141}
142
143#[cfg(test)]
144mod tests {
145    use super::*;
146
147    #[test]
148    fn get_packed_pipeline_matches_redis() {
149        for atomic in [false, true] {
150            let mut glide_pipe = pipe();
151            let mut redis_pipe = redis::pipe();
152
153            if atomic {
154                glide_pipe.atomic();
155                redis_pipe.atomic();
156            }
157
158            // Typed commands.
159            glide_pipe
160                .set("k", "v")
161                .incr("k", 1)
162                .ignore()
163                .get("k")
164                .rpush("l", &["a", "b"][..]);
165            redis_pipe
166                .set("k", "v")
167                .incr("k", 1)
168                .ignore()
169                .get("k")
170                .rpush("l", &["a", "b"][..]);
171
172            // Untyped command.
173            let big = vec![0xABu8; 4096];
174            glide_pipe
175                .cmd("APPEND")
176                .arg("k")
177                .arg(&big)
178                .arg(&["a", "b", "c"][..]);
179            redis_pipe
180                .cmd("APPEND")
181                .arg("k")
182                .arg(&big)
183                .arg(&["a", "b", "c"][..]);
184
185            assert_eq!(
186                glide_pipe.as_redis().get_packed_pipeline(),
187                redis_pipe.get_packed_pipeline(),
188                "glide pipeline encoding diverged from redis (atomic = {atomic})",
189            );
190        }
191    }
192}
glide/pipeline.rs

glide/
pipeline.rs
