ruststream_fred/pipeline/descriptors.rs
1//! The pipelined subscription sources under names of their own.
2//!
3//! A service does not build these: `.pipeline()` and `.atomic()` at the mount site produce them
4//! from the descriptor the attribute named. The names are what the builder's type and the
5//! compiler's messages show.
6
7use super::{Atomic, Pipelined, Plain};
8use crate::pubsub::{RedisPubSub, RedisPubSubPattern};
9use crate::{RedisList, RedisStream};
10
11/// A stream subscription with a window: what `.pipeline()` at the mount site makes of a [`RedisStream`].
12///
13/// # Examples
14///
15/// ```
16/// # mod demo {
17/// use ruststream_fred::stream::prelude::*;
18/// # #[derive(serde::Deserialize)]
19/// # struct Event { id: u64 }
20///
21/// #[subscriber(RedisStream::new("orders").group("workers"))]
22/// async fn record(event: &Event, Ctx(pipeline): Ctx<keys::Pipeline>) -> HandlerOutcome {
23/// if pipeline.incr(format!("seen:{}", event.id)).await.is_err() {
24/// return HandlerOutcome::retry();
25/// }
26/// HandlerOutcome::ack()
27/// }
28///
29/// fn app() -> RustStream {
30/// RustStream::new(AppInfo::new("events", "0.1.0")).with_broker(
31/// RedisBroker::standalone("redis://localhost:6379"),
32/// |b| {
33/// // The source of this registration is a `RedisStreamPipeline`.
34/// b.include(record.pipeline());
35/// },
36/// )
37/// }
38/// # }
39/// ```
40pub type RedisStreamPipeline = Pipelined<RedisStream, Plain>;
41
42/// A stream subscription with an atomic window: what `.pipeline().atomic()` at the mount site makes of a [`RedisStream`].
43///
44/// # Examples
45///
46/// ```
47/// # mod demo {
48/// use ruststream_fred::stream::prelude::*;
49/// # #[derive(serde::Deserialize)]
50/// # struct Event { id: u64 }
51///
52/// #[subscriber(RedisStream::new("{orders}").group("workers"))]
53/// async fn record(event: &Event, Ctx(pipeline): Ctx<keys::Pipeline>) -> HandlerOutcome {
54/// if pipeline.incr(format!("seen:{}", event.id)).await.is_err() {
55/// return HandlerOutcome::retry();
56/// }
57/// HandlerOutcome::ack()
58/// }
59///
60/// fn app() -> RustStream {
61/// RustStream::new(AppInfo::new("events", "0.1.0")).with_broker(
62/// RedisBroker::standalone("redis://localhost:6379"),
63/// |b| {
64/// // The source of this registration is a `RedisStreamAtomic`.
65/// b.include(record.pipeline().atomic());
66/// },
67/// )
68/// }
69/// # }
70/// ```
71pub type RedisStreamAtomic = Pipelined<RedisStream, Atomic>;
72
73/// A list subscription with a window: what `.pipeline()` at the mount site makes of a [`RedisList`].
74///
75/// # Examples
76///
77/// ```
78/// # mod demo {
79/// use ruststream_fred::list::prelude::*;
80/// # #[derive(serde::Deserialize)]
81/// # struct Event { id: u64 }
82///
83/// #[subscriber(RedisList::new("jobs").reliable())]
84/// async fn record(event: &Event, Ctx(pipeline): Ctx<keys::Pipeline>) -> HandlerOutcome {
85/// if pipeline.incr(format!("seen:{}", event.id)).await.is_err() {
86/// return HandlerOutcome::retry();
87/// }
88/// HandlerOutcome::ack()
89/// }
90///
91/// fn app() -> RustStream {
92/// RustStream::new(AppInfo::new("events", "0.1.0")).with_broker(
93/// RedisBroker::standalone("redis://localhost:6379"),
94/// |b| {
95/// // The source of this registration is a `RedisListPipeline`.
96/// b.include(record.pipeline());
97/// },
98/// )
99/// }
100/// # }
101/// ```
102pub type RedisListPipeline = Pipelined<RedisList, Plain>;
103
104/// A list subscription with an atomic window: what `.pipeline().atomic()` at the mount site makes of a [`RedisList`].
105///
106/// # Examples
107///
108/// ```
109/// # mod demo {
110/// use ruststream_fred::list::prelude::*;
111/// # #[derive(serde::Deserialize)]
112/// # struct Event { id: u64 }
113///
114/// #[subscriber(RedisList::new("{jobs}").reliable())]
115/// async fn record(event: &Event, Ctx(pipeline): Ctx<keys::Pipeline>) -> HandlerOutcome {
116/// if pipeline.incr(format!("seen:{}", event.id)).await.is_err() {
117/// return HandlerOutcome::retry();
118/// }
119/// HandlerOutcome::ack()
120/// }
121///
122/// fn app() -> RustStream {
123/// RustStream::new(AppInfo::new("events", "0.1.0")).with_broker(
124/// RedisBroker::standalone("redis://localhost:6379"),
125/// |b| {
126/// // The source of this registration is a `RedisListAtomic`.
127/// b.include(record.pipeline().atomic());
128/// },
129/// )
130/// }
131/// # }
132/// ```
133pub type RedisListAtomic = Pipelined<RedisList, Atomic>;
134
135/// A Pub/Sub channel subscription with a window: what `.pipeline()` at the mount site makes of a [`RedisPubSub`].
136///
137/// # Examples
138///
139/// ```
140/// # mod demo {
141/// use ruststream_fred::pubsub::prelude::*;
142/// # #[derive(serde::Deserialize)]
143/// # struct Event { id: u64 }
144///
145/// #[subscriber(RedisPubSub::new("events"))]
146/// async fn record(event: &Event, Ctx(pipeline): Ctx<keys::Pipeline>) -> HandlerOutcome {
147/// if pipeline.incr(format!("seen:{}", event.id)).await.is_err() {
148/// return HandlerOutcome::retry();
149/// }
150/// HandlerOutcome::ack()
151/// }
152///
153/// fn app() -> RustStream {
154/// RustStream::new(AppInfo::new("events", "0.1.0")).with_broker(
155/// RedisBroker::standalone("redis://localhost:6379"),
156/// |b| {
157/// // The source of this registration is a `RedisPubSubPipeline`.
158/// b.include(record.pipeline());
159/// },
160/// )
161/// }
162/// # }
163/// ```
164pub type RedisPubSubPipeline = Pipelined<RedisPubSub, Plain>;
165
166/// A Pub/Sub channel subscription with an atomic window: what `.pipeline().atomic()` at the mount site makes of a [`RedisPubSub`].
167///
168/// # Examples
169///
170/// ```
171/// # mod demo {
172/// use ruststream_fred::pubsub::prelude::*;
173/// # #[derive(serde::Deserialize)]
174/// # struct Event { id: u64 }
175///
176/// #[subscriber(RedisPubSub::new("events"))]
177/// async fn record(event: &Event, Ctx(pipeline): Ctx<keys::Pipeline>) -> HandlerOutcome {
178/// if pipeline.incr(format!("seen:{}", event.id)).await.is_err() {
179/// return HandlerOutcome::retry();
180/// }
181/// HandlerOutcome::ack()
182/// }
183///
184/// fn app() -> RustStream {
185/// RustStream::new(AppInfo::new("events", "0.1.0")).with_broker(
186/// RedisBroker::standalone("redis://localhost:6379"),
187/// |b| {
188/// // The source of this registration is a `RedisPubSubAtomic`.
189/// b.include(record.pipeline().atomic());
190/// },
191/// )
192/// }
193/// # }
194/// ```
195pub type RedisPubSubAtomic = Pipelined<RedisPubSub, Atomic>;
196
197/// A Pub/Sub pattern subscription with a window: what `.pipeline()` at the mount site makes of a [`RedisPubSubPattern`].
198///
199/// # Examples
200///
201/// ```
202/// # mod demo {
203/// use ruststream_fred::pubsub::prelude::*;
204/// # #[derive(serde::Deserialize)]
205/// # struct Event { id: u64 }
206///
207/// #[subscriber(RedisPubSubPattern::new("events.*"))]
208/// async fn record(event: &Event, Ctx(pipeline): Ctx<keys::Pipeline>) -> HandlerOutcome {
209/// if pipeline.incr(format!("seen:{}", event.id)).await.is_err() {
210/// return HandlerOutcome::retry();
211/// }
212/// HandlerOutcome::ack()
213/// }
214///
215/// fn app() -> RustStream {
216/// RustStream::new(AppInfo::new("events", "0.1.0")).with_broker(
217/// RedisBroker::standalone("redis://localhost:6379"),
218/// |b| {
219/// // The source of this registration is a `RedisPubSubPatternPipeline`.
220/// b.include(record.pipeline());
221/// },
222/// )
223/// }
224/// # }
225/// ```
226pub type RedisPubSubPatternPipeline = Pipelined<RedisPubSubPattern, Plain>;
227
228/// A Pub/Sub pattern subscription with an atomic window: what `.pipeline().atomic()` at the mount site makes of a [`RedisPubSubPattern`].
229///
230/// # Examples
231///
232/// ```
233/// # mod demo {
234/// use ruststream_fred::pubsub::prelude::*;
235/// # #[derive(serde::Deserialize)]
236/// # struct Event { id: u64 }
237///
238/// #[subscriber(RedisPubSubPattern::new("events.*"))]
239/// async fn record(event: &Event, Ctx(pipeline): Ctx<keys::Pipeline>) -> HandlerOutcome {
240/// if pipeline.incr(format!("seen:{}", event.id)).await.is_err() {
241/// return HandlerOutcome::retry();
242/// }
243/// HandlerOutcome::ack()
244/// }
245///
246/// fn app() -> RustStream {
247/// RustStream::new(AppInfo::new("events", "0.1.0")).with_broker(
248/// RedisBroker::standalone("redis://localhost:6379"),
249/// |b| {
250/// // The source of this registration is a `RedisPubSubPatternAtomic`.
251/// b.include(record.pipeline().atomic());
252/// },
253/// )
254/// }
255/// # }
256/// ```
257pub type RedisPubSubPatternAtomic = Pipelined<RedisPubSubPattern, Atomic>;