ruststream_fred/pipeline/
mod.rs

1//! Consumer-side pipelining: a subscription settles in a window, and a handler queues its own
2//! Redis commands into the delivery's segment of that window.
3//!
4//! - [pipeline][pipeline]: [`RedisPipelineSteps`], the `.pipeline()` step that opens the window.
5//! - [atomic][atomic]: `.atomic()`, one `MULTI` / `EXEC` per delivery.
6//! - [Commands][commands]: [`RedisPipeline`], the commands a handler queues through
7//!   [`keys::Pipeline`](crate::context::keys::Pipeline).
8//! - [bind][bind]: [`Bindable`], a slot's publishes in the delivery's round.
9//! - [InRound][in-round]: [`InRound`], the reply in the delivery's round.
10//! - [Flush][flush]: when a window leaves, and what each outcome does to what a handler queued.
11//!
12//! [pipeline]: https://powersemmi.github.io/ruststream-fred/latest/components/pipelining/
13//! [atomic]: https://powersemmi.github.io/ruststream-fred/latest/components/pipelining/atomic/
14//! [commands]: https://powersemmi.github.io/ruststream-fred/latest/components/pipelining/commands/
15//! [bind]: https://powersemmi.github.io/ruststream-fred/latest/components/pipelining/bind/
16//! [in-round]: https://powersemmi.github.io/ruststream-fred/latest/components/pipelining/in-round/
17//! [flush]: https://powersemmi.github.io/ruststream-fred/latest/components/pipelining/flush/
18
19mod commands;
20mod descriptors;
21mod forms;
22mod message;
23mod rounds;
24mod source;
25#[cfg(all(test, feature = "testing"))]
26mod tests;
27mod window;
28
29use std::fmt::{Debug, Formatter};
30use std::marker::PhantomData;
31use std::ops::Deref;
32
33use fred::bytes_utils::Str;
34use fred::clients::{Client, Pipeline};
35use fred::error::Error;
36use fred::interfaces::ClientLike;
37use fred::types::{ClusterHash, CustomCommand, MultipleKeys, MultipleValues, Value};
38use ruststream::StartAt;
39use ruststream::runtime::{
40    Declared, ForReply, Outgoing, PublishContext, PublishTransform, Reads, SubscriberBuilder,
41    SubscriberSettings,
42};
43
44use crate::partition::RedisPublishOptions;
45
46pub use descriptors::{
47    RedisListAtomic, RedisListPipeline, RedisPubSubAtomic, RedisPubSubPatternAtomic,
48    RedisPubSubPatternPipeline, RedisPubSubPipeline, RedisStreamAtomic, RedisStreamPipeline,
49};
50pub(crate) use forms::{ListForm, PubSubForm, StreamForm};
51pub use message::RoundMessage;
52pub(crate) use rounds::Rounds;
53pub use source::PipelinedSubscriber;
54pub(crate) use window::{Form, Owner, Round, Segment, Window};
55
56/// A subscription source with a window, as the `.pipeline()` mount step makes it.
57///
58/// The descriptor is a [`RedisStream`](crate::RedisStream), a [`RedisList`](crate::RedisList), a
59/// [`RedisPubSub`](crate::RedisPubSub) or a [`RedisPubSubPattern`](crate::RedisPubSubPattern).
60/// `Mode` is [`Plain`] until `.atomic()` makes it [`Atomic`]. A service does not build one: it
61/// names the descriptor in `#[subscriber(..)]` and chains [`pipeline`](RedisPipelineSteps::pipeline)
62/// where it mounts the handler. The type shows up in the builder's type, under the names
63/// [`RedisStreamPipeline`] and its siblings.
64///
65/// # Examples
66///
67/// ```
68/// # mod demo {
69/// use ruststream_fred::stream::prelude::*;
70/// # #[derive(serde::Deserialize)]
71/// # struct Order { id: u64 }
72///
73/// #[subscriber(RedisStream::new("orders").group("workers"))]
74/// async fn record(order: &Order, Ctx(pipeline): Ctx<keys::Pipeline>) -> HandlerOutcome {
75///     if pipeline.hset("orders:by-id", (order.id.to_string(), "seen")).await.is_err() {
76///         return HandlerOutcome::retry();
77///     }
78///     HandlerOutcome::ack()
79/// }
80///
81/// fn app() -> RustStream {
82///     RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
83///         RedisBroker::standalone("redis://localhost:6379"),
84///         |b| {
85///             b.include(record.pipeline());
86///         },
87///     )
88/// }
89/// # }
90/// ```
91#[must_use]
92pub struct Pipelined<Descriptor, Mode = Plain> {
93    descriptor: Descriptor,
94    mode: PhantomData<fn() -> Mode>,
95}
96
97impl<Descriptor: Clone, Mode> Clone for Pipelined<Descriptor, Mode> {
98    fn clone(&self) -> Self {
99        Self {
100            descriptor: self.descriptor.clone(),
101            mode: PhantomData,
102        }
103    }
104}
105
106impl<Descriptor: Debug, Mode: WindowMode> Debug for Pipelined<Descriptor, Mode> {
107    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
108        f.debug_struct("Pipelined")
109            .field("descriptor", &self.descriptor)
110            .field("atomic", &Mode::ATOMIC)
111            .finish()
112    }
113}
114
115impl<Descriptor, Mode> Pipelined<Descriptor, Mode> {
116    pub(crate) const fn wrap(descriptor: Descriptor) -> Self {
117        Self {
118            descriptor,
119            mode: PhantomData,
120        }
121    }
122
123    /// The descriptor the window was opened on.
124    pub(crate) const fn descriptor(&self) -> &Descriptor {
125        &self.descriptor
126    }
127
128    pub(crate) fn into_descriptor(self) -> Descriptor {
129        self.descriptor
130    }
131
132    /// The same window over the descriptor `step` returns: a base setting chained at the mount
133    /// site after `.pipeline()`.
134    pub(crate) fn map(self, step: impl FnOnce(Descriptor) -> Descriptor) -> Self {
135        Self::wrap(step(self.descriptor))
136    }
137}
138
139/// The window's mode before `.atomic()`: a segment is the handler's commands, in the window's
140/// pipeline.
141#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
142pub struct Plain;
143
144/// The window's mode after `.atomic()`: a segment is `MULTI`, the handler's commands, the settle,
145/// `EXEC`.
146#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
147pub struct Atomic;
148
149mod sealed {
150    pub trait Sealed {}
151    impl Sealed for super::Plain {}
152    impl Sealed for super::Atomic {}
153}
154
155/// The two modes of a window, [`Plain`] and [`Atomic`].
156///
157/// # Examples
158///
159/// ```
160/// # mod demo {
161/// use ruststream_fred::stream::prelude::*;
162/// # #[derive(serde::Deserialize)]
163/// # struct Order { id: u64 }
164///
165/// #[subscriber(RedisStream::new("{orders}").group("workers"))]
166/// async fn invoice(order: &Order, Ctx(pipeline): Ctx<keys::Pipeline>) -> HandlerOutcome {
167///     match pipeline.lpush("{orders}:invoices", order.id.to_string()).await {
168///         Ok(()) => HandlerOutcome::ack(),
169///         Err(_) => HandlerOutcome::retry(),
170///     }
171/// }
172///
173/// fn app() -> RustStream {
174///     RustStream::new(AppInfo::new("invoices", "0.1.0")).with_broker(
175///         RedisBroker::standalone("redis://localhost:6379"),
176///         |b| {
177///             // `.pipeline()` opens the window as `Plain`, and `.atomic()` makes it `Atomic`.
178///             b.include(invoice.pipeline().atomic());
179///         },
180///     )
181/// }
182/// # }
183/// ```
184pub trait WindowMode: sealed::Sealed + Send + Sync + 'static {
185    /// Whether a segment is wrapped in `MULTI` / `EXEC`.
186    const ATOMIC: bool;
187}
188
189impl WindowMode for Plain {
190    const ATOMIC: bool = false;
191}
192
193impl WindowMode for Atomic {
194    const ATOMIC: bool = true;
195}
196
197mod windowable {
198    /// The descriptors a window opens on, and nothing else.
199    #[diagnostic::on_unimplemented(
200        message = "`.pipeline()` opens a window on a `RedisStream`, `RedisList`, `RedisPubSub` or \
201                   `RedisPubSubPattern` subscription, and this one is `{Self}`",
202        label = "no window opens here",
203        note = "a subscription has one window: chain `.pipeline()` once, on the descriptor the \
204                attribute names"
205    )]
206    pub trait Base {}
207    impl Base for crate::RedisStream {}
208    impl Base for crate::RedisList {}
209    impl Base for crate::RedisPubSub {}
210    impl Base for crate::RedisPubSubPattern {}
211}
212
213/// A subscription source a window can open on: a descriptor of this crate, possibly under the
214/// core's `start_at(..)` wrapper.
215#[diagnostic::on_unimplemented(
216    message = "`.pipeline()` opens a window on a subscription of ruststream-fred, and `{Self}` is not one",
217    label = "no window opens on this source",
218    note = "name a `RedisStream`, `RedisList`, `RedisPubSub` or `RedisPubSubPattern` in \
219            `#[subscriber(..)]`, and chain `.pipeline()` once"
220)]
221pub trait OpensWindow {
222    /// The same source with a window.
223    type Pipelined;
224
225    /// Opens the window.
226    fn open_window(self) -> Self::Pipelined;
227}
228
229impl<Descriptor: windowable::Base> OpensWindow for Descriptor {
230    type Pipelined = Pipelined<Descriptor, Plain>;
231
232    fn open_window(self) -> Self::Pipelined {
233        Pipelined::wrap(self)
234    }
235}
236
237impl<Source: OpensWindow, Position> OpensWindow for StartAt<Source, Position> {
238    type Pipelined = StartAt<Source::Pipelined, Position>;
239
240    fn open_window(self) -> Self::Pipelined {
241        self.map_inner(OpensWindow::open_window)
242    }
243}
244
245/// A source that already has a window, which is what `.atomic()` needs.
246#[diagnostic::on_unimplemented(
247    message = "`.atomic()` needs `.pipeline()` first",
248    label = "this subscription has no pipeline",
249    note = "write `.pipeline().atomic()`: a transaction is one segment of the window's pipeline, \
250            and a subscription without a window has no pipeline to put it in"
251)]
252pub trait Windowed {
253    /// The same source with an atomic window.
254    type Atomic;
255
256    /// Makes the window atomic.
257    fn into_atomic(self) -> Self::Atomic;
258}
259
260impl<Descriptor> Windowed for Pipelined<Descriptor, Plain> {
261    type Atomic = Pipelined<Descriptor, Atomic>;
262
263    fn into_atomic(self) -> Self::Atomic {
264        Pipelined::wrap(self.descriptor)
265    }
266}
267
268impl<Source: Windowed, Position> Windowed for StartAt<Source, Position> {
269    type Atomic = StartAt<Source::Atomic, Position>;
270
271    fn into_atomic(self) -> Self::Atomic {
272        self.map_inner(Windowed::into_atomic)
273    }
274}
275
276/// The builder step behind [`RedisPipelineSteps::pipeline`].
277#[diagnostic::on_unimplemented(
278    message = "`.pipeline()` opens a window on a subscription of ruststream-fred, and this \
279               registration is not one",
280    label = "no window opens on this registration",
281    note = "name a `RedisStream`, `RedisList`, `RedisPubSub` or `RedisPubSubPattern` in \
282            `#[subscriber(..)]`, and chain `.pipeline()` once"
283)]
284pub trait PipelineStep {
285    /// The builder over the pipelined source.
286    type Out;
287
288    /// Opens the window on the builder's source.
289    fn apply_pipeline(self) -> Self::Out;
290}
291
292impl<Def, Source, State, DefCodec> PipelineStep for SubscriberBuilder<Def, Source, State, DefCodec>
293where
294    Def: Declared,
295    Source: OpensWindow,
296{
297    type Out = SubscriberBuilder<Def, Source::Pipelined, State, DefCodec>;
298
299    fn apply_pipeline(self) -> Self::Out {
300        self.map_source(OpensWindow::open_window)
301    }
302}
303
304/// The builder step behind [`RedisPipelineSteps::atomic`].
305#[diagnostic::on_unimplemented(
306    message = "`.atomic()` needs `.pipeline()` first",
307    label = "this subscription has no pipeline",
308    note = "write `.pipeline().atomic()`: a transaction is one segment of the window's pipeline, \
309            and a subscription without a window has no pipeline to put it in"
310)]
311pub trait AtomicStep {
312    /// The builder over the source with an atomic window.
313    type Out;
314
315    /// Makes the window atomic.
316    fn apply_atomic(self) -> Self::Out;
317}
318
319impl<Def, Source, State, DefCodec> AtomicStep for SubscriberBuilder<Def, Source, State, DefCodec>
320where
321    Def: Declared,
322    Source: Windowed,
323{
324    type Out = SubscriberBuilder<Def, Source::Atomic, State, DefCodec>;
325
326    fn apply_atomic(self) -> Self::Out {
327        self.map_source(Windowed::into_atomic)
328    }
329}
330
331/// The window steps of a mount site: `.pipeline()` opens a consumer window on the subscription,
332/// and `.atomic()` after it makes each delivery's segment one `MULTI` / `EXEC`.
333///
334/// The steps chain on the handler where it is mounted, like the core's settings, after the
335/// descriptor the attribute named and after any base setting such as
336/// [`block`](crate::RedisSubscribeExt::block). They apply to a [`RedisStream`](crate::RedisStream),
337/// a [`RedisList`](crate::RedisList), a [`RedisPubSub`](crate::RedisPubSub) and a
338/// [`RedisPubSubPattern`](crate::RedisPubSubPattern) alike; see [the module](self) for what the
339/// window does.
340///
341/// Under `.atomic()` a delivery's segment is `MULTI`, the commands its handler queued, the
342/// delivery's settle where the form has one, and `EXEC`, sent inside the window's pipeline, so the
343/// handler's Redis side effects and its acknowledgement happen together or not at all. It buys no
344/// rollback: Redis has none, and a command that fails inside `EXEC` (a `WRONGTYPE`, say) leaves
345/// the others executed.
346///
347/// On a cluster, Redis runs a transaction only when every key of it hashes to one slot. A
348/// segment's slot is the subscription key's, so a handler's keys share it through a hash tag
349/// (`{orders}:invoices` beside a stream `{orders}`). A command for another slot is answered with a
350/// redirection inside the transaction, and Redis then refuses the whole `EXEC`, so a segment is
351/// never executed in part.
352///
353/// A handler that reads `Ctx<keys::Pipeline>` needs `.pipeline()` at its mount site. Without it
354/// the mount does not compile, and the error is the framework's general one about the handler's
355/// context not matching the subscription's (`PipelineContext: BuildContext<RedisMessage>`).
356///
357/// # Examples
358///
359/// ```
360/// # mod demo {
361/// use ruststream_fred::stream::prelude::*;
362/// # #[derive(serde::Deserialize)]
363/// # struct Order { id: u64 }
364///
365/// #[subscriber(RedisStream::new("{orders}").group("workers"))]
366/// async fn invoice(order: &Order, Ctx(pipeline): Ctx<keys::Pipeline>) -> HandlerOutcome {
367///     if pipeline.lpush("{orders}:invoices", order.id.to_string()).await.is_err() {
368///         return HandlerOutcome::retry();
369///     }
370///     HandlerOutcome::ack()
371/// }
372///
373/// fn app() -> RustStream {
374///     RustStream::new(AppInfo::new("invoices", "0.1.0")).with_broker(
375///         RedisBroker::standalone("redis://localhost:6379"),
376///         |b| {
377///             b.include(invoice.pipeline().atomic());
378///         },
379///     )
380/// }
381/// # }
382/// ```
383///
384/// A subscription without a window has nothing to make atomic:
385///
386/// ```compile_fail
387/// # mod demo {
388/// use ruststream_fred::stream::prelude::*;
389/// # #[derive(serde::Deserialize)]
390/// # struct Order { id: u64 }
391///
392/// #[subscriber(RedisStream::new("orders").group("workers"))]
393/// async fn invoice(order: &Order) -> HandlerOutcome {
394///     let _ = order.id;
395///     HandlerOutcome::ack()
396/// }
397///
398/// fn app() -> RustStream {
399///     RustStream::new(AppInfo::new("invoices", "0.1.0")).with_broker(
400///         RedisBroker::standalone("redis://localhost:6379"),
401///         |b| {
402///             b.include(invoice.atomic());
403///         },
404///     )
405/// }
406/// # }
407/// ```
408pub trait RedisPipelineSteps: Declared {
409    /// Opens a consumer window on the subscription: its settles and the commands its handlers
410    /// queue through [`keys::Pipeline`](crate::context::keys::Pipeline) leave together, in one
411    /// pipeline on a connection of the pool.
412    ///
413    /// On a stream the window flushes once the read's `COUNT` has settled, once nothing is
414    /// outstanding, or when the subscription stops: under load one `XACK` per fetched batch. A
415    /// pipelined list reads in batches, and a reliable list's `LREM` settles ride the window.
416    /// Pub/Sub, on a channel or a pattern, settles nothing, so its window carries the handlers'
417    /// commands alone, and leaves once nothing is in flight.
418    ///
419    /// # Examples
420    ///
421    /// ```
422    /// # mod demo {
423    /// use ruststream_fred::stream::prelude::*;
424    /// use serde::{Deserialize, Serialize};
425    ///
426    /// #[derive(Deserialize)]
427    /// struct Order {
428    ///     id: u64,
429    /// }
430    ///
431    /// #[derive(Serialize, Outgoing)]
432    /// #[outgoing(name = "receipts")]
433    /// struct Receipt {
434    ///     id: u64,
435    /// }
436    ///
437    /// #[subscriber(RedisStream::new("orders").group("workers"), reply)]
438    /// async fn record(
439    ///     order: &Order,
440    ///     Ctx(pipeline): Ctx<keys::Pipeline>,
441    /// ) -> Result<Receipt, HandlerOutcome> {
442    ///     // Queued: it runs after the handler returns, with the delivery's `XACK`.
443    ///     if pipeline.hincrby("orders:count", "seen", 1).await.is_err() {
444    ///         return Err(HandlerOutcome::retry());
445    ///     }
446    ///     Ok(Receipt { id: order.id })
447    /// }
448    ///
449    /// #[ruststream::app]
450    /// fn app() -> impl App {
451    ///     RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
452    ///         RedisBroker::standalone("redis://localhost:6379"),
453    ///         |b| {
454    ///             // The reply joins the delivery's round and leaves after what the handler queued.
455    ///             b.include(record.pipeline()).out_reply(Publish).transform(InRound);
456    ///         },
457    ///     )
458    /// }
459    /// # }
460    /// # fn main() {}
461    /// ```
462    fn pipeline(self) -> <Self::Settings as PipelineStep>::Out
463    where
464        Self::Settings: PipelineStep,
465    {
466        self.declare().apply_pipeline()
467    }
468
469    /// Makes each delivery's segment one `MULTI` / `EXEC`. Follows `.pipeline()`.
470    fn atomic(self) -> <Self::Settings as AtomicStep>::Out
471    where
472        Self::Settings: AtomicStep,
473    {
474        self.declare().apply_atomic()
475    }
476}
477
478impl<D: Declared> RedisPipelineSteps for D {}
479
480mod bindable {
481    /// What a publisher of this crate is named by in a binding.
482    pub trait Named {
483        fn round_name(&self) -> u64;
484    }
485}
486
487/// A publisher of this crate, which `pipeline.bind(out)` can bind to a delivery's round.
488///
489/// Implemented by every publisher this crate's policies pair into, and by nothing else: a
490/// publisher of another broker publishes through its own connection, which no round of this one
491/// can carry. A handler bounds its slot with it, `Out(out): Out<impl Bindable>`, to bind it.
492///
493/// # Examples
494///
495/// ```
496/// # mod demo {
497/// use ruststream_fred::stream::prelude::*;
498/// use serde::Serialize;
499/// # #[derive(serde::Deserialize)]
500/// # struct Order { id: u64 }
501/// #[derive(Serialize, Outgoing)]
502/// #[outgoing(name = "audit")]
503/// struct Audit {
504///     id: u64,
505/// }
506///
507/// /// The audit entry leaves with the delivery's acknowledgement, and not without it.
508/// #[subscriber(RedisStream::new("orders").group("workers"))]
509/// async fn record(
510///     order: &Order,
511///     Ctx(pipeline): Ctx<keys::Pipeline>,
512///     Out(out): Out<impl Bindable>,
513/// ) -> HandlerOutcome {
514///     let out = pipeline.bind(out);
515///     if out.message(&Audit { id: order.id }).publish().await.is_err() {
516///         return HandlerOutcome::retry();
517///     }
518///     HandlerOutcome::ack()
519/// }
520///
521/// fn app() -> RustStream {
522///     RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
523///         RedisBroker::standalone("redis://localhost:6379"),
524///         |b| {
525///             b.include(record.pipeline()).out(DefaultSlot, Publish).build();
526///         },
527///     )
528/// }
529/// # }
530/// ```
531///
532/// A publisher of another transport has no round to join, so binding one does not compile:
533///
534/// ```compile_fail,E0277
535/// use std::convert::Infallible;
536///
537/// use ruststream::{Lend, OutgoingMessage, Publisher};
538/// use ruststream_fred::pipeline::Bindable;
539///
540/// struct Elsewhere;
541///
542/// impl Publisher for Elsewhere {
543///     type Payload = Lend;
544///     type Error = Infallible;
545///     type Options = ();
546///
547///     async fn publish(
548///         &self,
549///         _msg: OutgoingMessage<'_, &[u8]>,
550///         _options: Option<&()>,
551///     ) -> Result<(), Infallible> {
552///         Ok(())
553///     }
554/// }
555///
556/// fn bindable<P: Bindable>() {}
557///
558/// bindable::<Elsewhere>();
559/// ```
560#[diagnostic::on_unimplemented(
561    message = "`{Self}` is not a publisher of ruststream-fred, so it cannot join a delivery's round",
562    label = "this slot publishes through another broker",
563    note = "bind a slot whose policy is one of this crate's (`stream::Publish`, `list::Publish`, \
564            `pubsub::Publish`), and bound the slot parameter as `Out<impl Bindable>`"
565)]
566pub trait Bindable: ruststream::Publisher + bindable::Named {}
567
568impl<T: ruststream::Publisher + bindable::Named> Bindable for T {}
569
570impl bindable::Named for crate::RedisPublisher {
571    fn round_name(&self) -> u64 {
572        self.round_name()
573    }
574}
575
576impl bindable::Named for crate::RedisListPublisher {
577    fn round_name(&self) -> u64 {
578        self.round_name()
579    }
580}
581
582impl bindable::Named for crate::RedisPubSubPublisher {
583    fn round_name(&self) -> u64 {
584        self.round_name()
585    }
586}
587
588impl bindable::Named for crate::RedisDefaultPublisher {
589    fn round_name(&self) -> u64 {
590        self.round_name()
591    }
592}
593
594/// The reply transform that puts a reply in the round of the delivery it answers.
595///
596/// A reply is published after the handler returns and before the delivery settles. Mounted on
597/// the reply position of a `.pipeline()` subscription, this transform queues it into the
598/// delivery's segment instead: it leaves with the window, after what the handler queued, and
599/// under `.atomic()` inside the delivery's `MULTI` / `EXEC`. On a subscription without a window
600/// the reply leaves at once, as it does without the transform.
601///
602/// # Examples
603///
604/// ```
605/// # mod demo {
606/// use ruststream_fred::stream::prelude::*;
607/// use serde::Serialize;
608/// # #[derive(serde::Deserialize)]
609/// # struct Order { id: u64 }
610/// #[derive(Serialize, Outgoing)]
611/// #[outgoing(name = "receipts")]
612/// struct Receipt {
613///     id: u64,
614/// }
615///
616/// #[subscriber(RedisStream::new("orders").group("workers"), reply)]
617/// async fn issue(order: &Order) -> Receipt {
618///     Receipt { id: order.id }
619/// }
620///
621/// fn app() -> RustStream {
622///     RustStream::new(AppInfo::new("receipts", "0.1.0")).with_broker(
623///         RedisBroker::standalone("redis://localhost:6379"),
624///         |b| {
625///             b.include(issue.pipeline()).out_reply(Publish).transform(InRound);
626///         },
627///     )
628/// }
629/// # }
630/// ```
631#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
632pub struct InRound;
633
634impl<C> PublishTransform<ForReply<C>, RedisPublishOptions> for InRound {
635    type Destination = Reads;
636
637    fn apply(
638        &self,
639        _out: &mut Outgoing<'_>,
640        options: &mut Option<RedisPublishOptions>,
641        _cx: &PublishContext<'_, C>,
642    ) {
643        options
644            .get_or_insert_with(RedisPublishOptions::default)
645            .join_round = true;
646    }
647}
648
649/// The delivery's own buffer in the window: what a handler queues its Redis commands into.
650///
651/// Reached as `Ctx<keys::Pipeline>` on a `.pipeline()` subscription. It carries `fred`'s command
652/// methods, each queueing the command and returning `Ok(())` once it is queued: the command runs
653/// after the handler has returned, and only if the delivery is acknowledged, so its reply is not
654/// available inside the handler. A command queued before `ack` executes before that `ack`. The
655/// window sends it; there is no method here that sends. For a command whose answer the handler needs now, take
656/// `Ctx<keys::FredPool>`.
657///
658/// The buffer is created on the first queued command, so a delivery whose handler queues nothing
659/// costs the window nothing.
660///
661/// # Examples
662///
663/// ```
664/// # mod demo {
665/// use ruststream_fred::stream::prelude::*;
666/// # #[derive(serde::Deserialize)]
667/// # struct Order { id: u64 }
668///
669/// /// Records the order and acknowledges it in one round trip with the rest of the window.
670/// #[subscriber(RedisStream::new("orders").group("workers"))]
671/// async fn record(order: &Order, Ctx(pipeline): Ctx<keys::Pipeline>) -> HandlerOutcome {
672///     if pipeline.hset("orders:by-id", (order.id.to_string(), "seen")).await.is_err() {
673///         return HandlerOutcome::retry();
674///     }
675///     HandlerOutcome::ack()
676/// }
677///
678/// fn app() -> RustStream {
679///     RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
680///         RedisBroker::standalone("redis://localhost:6379"),
681///         |b| {
682///             b.include(record.pipeline());
683///         },
684///     )
685/// }
686/// # }
687/// ```
688///
689/// A mount site without `.pipeline()` has no round to hand out, so the mount does not compile:
690///
691/// ```compile_fail
692/// # mod demo {
693/// use ruststream::prelude::*;
694/// use ruststream::subscriber;
695/// use ruststream_fred::context::keys;
696/// use ruststream_fred::RedisStream;
697/// # #[derive(serde::Deserialize)]
698/// # struct Order { id: u64 }
699///
700/// #[subscriber(RedisStream::new("orders").group("workers"))]
701/// async fn record(order: &Order, Ctx(pipeline): Ctx<keys::Pipeline>) -> HandlerOutcome {
702///     let _ = (order, pipeline);
703///     HandlerOutcome::ack()
704/// }
705///
706/// fn app() -> RustStream {
707///     RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
708///         ruststream_fred::RedisBroker::standalone("redis://localhost:6379"),
709///         |b| {
710///             b.include(record);
711///         },
712///     )
713/// }
714/// # }
715/// ```
716#[derive(Clone)]
717pub struct RedisPipeline {
718    round: Round,
719}
720
721impl Debug for RedisPipeline {
722    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
723        f.debug_struct("RedisPipeline").finish_non_exhaustive()
724    }
725}
726
727impl RedisPipeline {
728    pub(crate) const fn new(round: Round) -> Self {
729        Self { round }
730    }
731
732    /// The delivery's segment, created on first use.
733    async fn segment(&self) -> Result<Segment, Error> {
734        self.round.segment().await
735    }
736
737    /// Binds a slot's publisher to this delivery's round, and hands the slot back.
738    ///
739    /// What the slot publishes from this delivery's handler is queued into the delivery's segment
740    /// from then on: it leaves with the window, after what was queued before it, only if the
741    /// delivery is acknowledged, and under `.atomic()` inside the delivery's `MULTI` / `EXEC`.
742    /// The slot keeps its codec and its transforms. A slot publish the handler does not bind
743    /// leaves at once.
744    ///
745    /// See [`Bindable`] for a handler that binds its slot.
746    ///
747    /// # Examples
748    ///
749    /// ```
750    /// # mod demo {
751    /// use ruststream_fred::stream::prelude::*;
752    /// use serde::Serialize;
753    /// # #[derive(serde::Deserialize)]
754    /// # struct Order { id: u64 }
755    /// #[derive(Serialize, Outgoing)]
756    /// #[outgoing(name = "audit")]
757    /// struct Audit {
758    ///     id: u64,
759    /// }
760    ///
761    /// #[subscriber(RedisStream::new("orders").group("workers"))]
762    /// async fn record(
763    ///     order: &Order,
764    ///     Ctx(pipeline): Ctx<keys::Pipeline>,
765    ///     Out(out): Out<impl Bindable>,
766    /// ) -> HandlerOutcome {
767    ///     // Queued into the delivery's segment: the audit entry leaves only with the `XACK`.
768    ///     match pipeline.bind(out).message(&Audit { id: order.id }).publish().await {
769    ///         Ok(()) => HandlerOutcome::ack(),
770    ///         Err(_) => HandlerOutcome::retry(),
771    ///     }
772    /// }
773    /// #
774    /// # fn app() -> RustStream {
775    /// #     RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
776    /// #         RedisBroker::standalone("redis://localhost:6379"),
777    /// #         |b| {
778    /// #             b.include(record.pipeline()).out(DefaultSlot, Publish).build();
779    /// #         },
780    /// #     )
781    /// # }
782    /// # }
783    /// ```
784    pub fn bind<'o, O>(&self, out: &'o O) -> &'o O
785    where
786        O: Deref,
787        O::Target: Bindable,
788    {
789        self.round.bind(bindable::Named::round_name(&**out));
790        out
791    }
792
793    /// Queues a Lua script run by its SHA1 digest (`EVALSHA`), with its keys and arguments.
794    ///
795    /// Queued into this delivery's segment; it runs after the handler returns, when the delivery
796    /// is acknowledged. On a cluster the command goes to the node of the first key.
797    ///
798    /// # Errors
799    ///
800    /// Returns `fred`'s error when the arguments do not convert, or when the delivery has already
801    /// settled.
802    ///
803    /// # Examples
804    ///
805    /// ```
806    /// # mod demo {
807    /// use ruststream_fred::stream::prelude::*;
808    /// # #[derive(serde::Deserialize)]
809    /// # struct Order { id: u64 }
810    ///
811    /// /// The digest of a script the service loaded with `SCRIPT LOAD` at startup.
812    /// const MARK_SEEN: &str = "a42059b356c875f0717db19a51f6aaca9ae659ea";
813    ///
814    /// #[subscriber(RedisStream::new("orders").group("workers"))]
815    /// async fn record(order: &Order, Ctx(pipeline): Ctx<keys::Pipeline>) -> HandlerOutcome {
816    ///     let id = order.id.to_string();
817    ///     match pipeline.evalsha(MARK_SEEN, vec!["orders:seen"], vec![id]).await {
818    ///         Ok(()) => HandlerOutcome::ack(),
819    ///         Err(_) => HandlerOutcome::retry(),
820    ///     }
821    /// }
822    /// #
823    /// # fn app() -> RustStream {
824    /// #     RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
825    /// #         RedisBroker::standalone("redis://localhost:6379"),
826    /// #         |b| {
827    /// #             b.include(record.pipeline());
828    /// #         },
829    /// #     )
830    /// # }
831    /// # }
832    /// ```
833    pub async fn evalsha<S, K, V>(&self, hash: S, keys: K, args: V) -> Result<(), Error>
834    where
835        S: Into<Str> + Send,
836        K: Into<MultipleKeys> + Send,
837        V: TryInto<MultipleValues> + Send,
838        V::Error: Into<Error> + Send,
839    {
840        let args = args.try_into().map_err(Into::into)?;
841        self.script("EVALSHA", hash.into(), keys.into(), args).await
842    }
843
844    /// Queues a Lua script run by its source (`EVAL`), with its keys and arguments.
845    ///
846    /// Queued into this delivery's segment; it runs after the handler returns, when the delivery
847    /// is acknowledged. On a cluster the command goes to the node of the first key.
848    ///
849    /// # Errors
850    ///
851    /// Returns `fred`'s error when the arguments do not convert, or when the delivery has already
852    /// settled.
853    ///
854    /// # Examples
855    ///
856    /// ```
857    /// # mod demo {
858    /// use ruststream_fred::stream::prelude::*;
859    /// # #[derive(serde::Deserialize)]
860    /// # struct Order { id: u64 }
861    ///
862    /// const MARK_SEEN: &str = "return redis.call('SADD', KEYS[1], ARGV[1])";
863    ///
864    /// #[subscriber(RedisStream::new("orders").group("workers"))]
865    /// async fn record(order: &Order, Ctx(pipeline): Ctx<keys::Pipeline>) -> HandlerOutcome {
866    ///     let id = order.id.to_string();
867    ///     match pipeline.eval(MARK_SEEN, vec!["orders:seen"], vec![id]).await {
868    ///         Ok(()) => HandlerOutcome::ack(),
869    ///         Err(_) => HandlerOutcome::retry(),
870    ///     }
871    /// }
872    /// #
873    /// # fn app() -> RustStream {
874    /// #     RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
875    /// #         RedisBroker::standalone("redis://localhost:6379"),
876    /// #         |b| {
877    /// #             b.include(record.pipeline());
878    /// #         },
879    /// #     )
880    /// # }
881    /// # }
882    /// ```
883    pub async fn eval<S, K, V>(&self, script: S, keys: K, args: V) -> Result<(), Error>
884    where
885        S: Into<Str> + Send,
886        K: Into<MultipleKeys> + Send,
887        V: TryInto<MultipleValues> + Send,
888        V::Error: Into<Error> + Send,
889    {
890        let args = args.try_into().map_err(Into::into)?;
891        self.script("EVAL", script.into(), keys.into(), args).await
892    }
893
894    /// A script run as `fred` sends one: the script or its digest, the key count, the keys and
895    /// the arguments, routed by the first key.
896    async fn script(
897        &self,
898        command: &'static str,
899        body: Str,
900        keys: MultipleKeys,
901        args: Value,
902    ) -> Result<(), Error> {
903        let keys = keys.inner();
904        let hash = keys.first().map_or(ClusterHash::Random, |key| {
905            ClusterHash::Custom(key.cluster_hash())
906        });
907        let mut values = Vec::with_capacity(2 + keys.len());
908        values.push(Value::from(body));
909        values.push(Value::from(i64::try_from(keys.len()).unwrap_or(i64::MAX)));
910        values.extend(keys.into_iter().map(Value::from));
911        match args {
912            Value::Array(args) => values.extend(args),
913            Value::Null => {}
914            arg => values.push(arg),
915        }
916        self.custom(CustomCommand::new_static(command, hash, false), values)
917            .await
918    }
919
920    /// Queues a command this facade does not name, by its name and its arguments.
921    ///
922    /// Queued into this delivery's segment; it runs after the handler returns, when the delivery
923    /// is acknowledged.
924    ///
925    /// # Errors
926    ///
927    /// Returns `fred`'s error when the arguments do not convert, or when the delivery has already
928    /// settled.
929    ///
930    /// # Examples
931    ///
932    /// ```
933    /// # mod demo {
934    /// use fred::types::{ClusterHash, CustomCommand};
935    /// use ruststream_fred::stream::prelude::*;
936    /// # #[derive(serde::Deserialize)]
937    /// # struct Order { id: u64 }
938    ///
939    /// #[subscriber(RedisStream::new("orders").group("workers"))]
940    /// async fn record(order: &Order, Ctx(pipeline): Ctx<keys::Pipeline>) -> HandlerOutcome {
941    ///     // Refreshes the order's last-access time, so an LRU eviction keeps it.
942    ///     let touch = CustomCommand::new_static("TOUCH", ClusterHash::FirstKey, false);
943    ///     match pipeline.custom(touch, vec![format!("order:{}", order.id)]).await {
944    ///         Ok(()) => HandlerOutcome::ack(),
945    ///         Err(_) => HandlerOutcome::retry(),
946    ///     }
947    /// }
948    /// #
949    /// # fn app() -> RustStream {
950    /// #     RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
951    /// #         RedisBroker::standalone("redis://localhost:6379"),
952    /// #         |b| {
953    /// #             b.include(record.pipeline());
954    /// #         },
955    /// #     )
956    /// # }
957    /// # }
958    /// ```
959    pub async fn custom<T>(&self, command: CustomCommand, args: Vec<T>) -> Result<(), Error>
960    where
961        T: TryInto<Value> + Send,
962        T::Error: Into<Error> + Send,
963    {
964        match self.segment().await? {
965            Segment::Plain(segment) => segment.custom::<(), T>(command, args).await,
966            Segment::Atomic(segment) => segment.custom::<(), T>(command, args).await,
967        }
968    }
969}
970
971/// A fresh `fred` pipeline on `client`, for one segment or one flush's settles.
972pub(crate) fn pipeline_on(client: &Client) -> Pipeline<Client> {
973    client.pipeline()
974}
ruststream_fred/pipeline/mod.rs

ruststream_fred/pipeline/
mod.rs
