ruststream_fred/
settings.rs

1//! Redis subscription options at the mount site.
2//!
3//! The framework passes exactly one number down to a broker's subscriber: the batch size, named by
4//! the core's own `batch(n)` step. Everything else about how a read forms is this crate's
5//! vocabulary, and [`RedisSubscribeExt`] is where it chains - after the size, on the same line.
6//!
7//! There is deliberately no publisher-side twin over the core's `MapPublisher` hook. A subscription
8//! needs one because its source is written inside `#[subscriber(..)]`, so a mount site has no other
9//! way to reach `block(..)`; a publish policy is constructed in the `.out(marker, policy)` call
10//! itself, where every option it carries (`RedisPubSubPublish::mode`, `RedisListPublish::ttl`) is
11//! already one chain away. A twin would be a second spelling for the same setting, and its
12//! envelope-framing method would be shadowed by the chain's own `.codec(..)`, which encodes the
13//! reply rather than the frame around it.
14
15use std::time::Duration;
16
17use ruststream::runtime::{Declared, SubscriberBuilder, SubscriberSettings};
18
19use crate::list::RedisList;
20use crate::pipeline::Pipelined;
21use crate::stream::RedisStream;
22
23/// The Redis subscription options a mount site names, chained after the core's settings steps.
24///
25/// Import it (or glob a form's prelude) to reach [`block`](Self::block) on a builder over a
26/// [`RedisStream`] or a [`RedisList`], with or without the window `.pipeline()` opens. The trait
27/// is bound to those sources, so the method does not appear on a builder for another broker or
28/// for a form that has no blocking read.
29///
30/// # Examples
31///
32/// ```
33/// # mod demo {
34/// use std::time::Duration;
35///
36/// use ruststream_fred::stream::prelude::*;
37/// # use serde::Deserialize;
38/// # #[derive(Deserialize)]
39/// # struct Order {
40/// #     amount: u64,
41/// # }
42///
43/// #[subscriber(RedisStream::new("orders").group("workers"))]
44/// async fn bill(orders: &[Order]) -> HandlerOutcome {
45///     let total: u64 = orders.iter().map(|order| order.amount).sum();
46///     println!("billed {} orders, {total} in all", orders.len());
47///     HandlerOutcome::ack()
48/// }
49///
50/// #[ruststream::app]
51/// fn app() -> impl App {
52///     RustStream::new(AppInfo::new("billing", "0.1.0")).with_broker(
53///         RedisBroker::standalone("redis://localhost:6379"),
54///         |b| {
55///             // The size is the core's word, the block is Redis's, and they read in that order.
56///             b.include(bill.batch(nonzero!(6)).block(Duration::from_secs(5)));
57///         },
58///     )
59/// }
60/// # }
61/// ```
62pub trait RedisSubscribeExt {
63    /// How long one read blocks waiting for entries, overriding what the descriptor named.
64    ///
65    /// On a stream this is the `XREADGROUP` server-side `BLOCK` (and, in reclaim mode, the poll
66    /// interval between empty `XAUTOCLAIM` scans); on a list it is the `BRPOP` / `BLMOVE` timeout.
67    /// Both default to five seconds.
68    #[must_use]
69    fn block(self, block: Duration) -> Self;
70}
71
72impl<Def, State, DefCodec> RedisSubscribeExt
73    for SubscriberBuilder<Def, RedisStream, State, DefCodec>
74where
75    Def: Declared,
76{
77    fn block(self, block: Duration) -> Self {
78        self.map_source(|source| source.block(block))
79    }
80}
81
82impl<Def, State, DefCodec> RedisSubscribeExt for SubscriberBuilder<Def, RedisList, State, DefCodec>
83where
84    Def: Declared,
85{
86    fn block(self, block: Duration) -> Self {
87        self.map_source(|source| source.block(block))
88    }
89}
90
91/// The same option on a stream whose window `.pipeline()` has opened, so the base settings read
92/// in either order around it.
93impl<Def, Mode, State, DefCodec> RedisSubscribeExt
94    for SubscriberBuilder<Def, Pipelined<RedisStream, Mode>, State, DefCodec>
95where
96    Def: Declared,
97{
98    fn block(self, block: Duration) -> Self {
99        self.map_source(|source| source.map(|stream| stream.block(block)))
100    }
101}
102
103/// The same option on a list whose window `.pipeline()` has opened.
104impl<Def, Mode, State, DefCodec> RedisSubscribeExt
105    for SubscriberBuilder<Def, Pipelined<RedisList, Mode>, State, DefCodec>
106where
107    Def: Declared,
108{
109    fn block(self, block: Duration) -> Self {
110        self.map_source(|source| source.map(|list| list.block(block)))
111    }
112}
ruststream_fred/settings.rs

ruststream_fred/
settings.rs
