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

ruststream_fred/pipeline/
mod.rs
