glide/sync/
pipeline.rs

1// Copyright Valkey GLIDE Project Contributors - SPDX Identifier: Apache-2.0
2//! Pipeline dispatch and typed pipeline execution on sync clients.
3
4use crate::ValkeyResult;
5use crate::pipeline::Pipeline;
6use crate::value::FromValkeyValue;
7
8/// Runs a pipeline of commands on a sync client.
9/// Implemented by [`SyncGlideClient`] and [`SyncGlideClusterClient`].
10#[doc(hidden)]
11#[sealed::sealed(pub(crate))]
12pub trait SyncPipelineDispatch {
13    fn glide_dispatch_pipeline<T: FromValkeyValue + Send>(
14        &self,
15        pipeline: &Pipeline,
16    ) -> ValkeyResult<T>;
17}
18
19/// Extension for running a [`Pipeline`] on a GLIDE sync client with
20/// with typed decoding and **zero extra payload copies**.
21///
22/// Like the rest of the sync layer, this blocks on the internal runtime and
23/// therefore **must not be called from within an async context** (doing so
24/// panics with tokio's "cannot block the current thread from within a runtime"
25/// — use the async [`crate::PipelineExt::query_async`] there instead).
26///
27/// ```rust,no_run
28/// use glide::sync::{PipelineExt, SyncGlideClient};
29/// # fn demo(client: &SyncGlideClient) -> glide::ValkeyResult<()> {
30/// let (a, b): (i64, i64) = glide::pipe()
31///     .atomic()
32///     .incr("c", 1)
33///     .incr("c", 1)
34///     .query(client)?;
35/// # let _ = (a, b); Ok(()) }
36/// ```
37pub trait PipelineExt {
38    /// Execute this pipeline on a blocking GLIDE client.
39    ///
40    /// Errors abort with the first errored command's reply. Ignored commands are
41    /// dropped before decoding, and an atomic pipeline's `EXEC` reply is unwrapped.
42    ///
43    /// Mirrors `redis-rs`'s `query`.
44    fn query<T: FromValkeyValue + Send>(&self, con: &impl SyncPipelineDispatch) -> ValkeyResult<T>;
45
46    /// Execute this pipeline on a blocking GLIDE client, discarding the replies.
47    ///
48    /// Mirrors `redis-rs`'s `exec`.
49    fn exec(&self, con: &impl SyncPipelineDispatch) -> ValkeyResult<()> {
50        self.query(con)
51    }
52}
53
54impl PipelineExt for Pipeline {
55    fn query<T: FromValkeyValue + Send>(&self, con: &impl SyncPipelineDispatch) -> ValkeyResult<T> {
56        con.glide_dispatch_pipeline(self)
57    }
58}
glide/sync/pipeline.rs

glide/sync/
pipeline.rs
