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
19#[cfg(doc)]
20use crate::list::RedisList;
21#[cfg(doc)]
22use crate::stream::RedisStream;
23
24/// The Redis subscription options a mount site names, chained after the core's settings steps.
25///
26/// Import it (or glob a form's prelude) to reach [`block`](Self::block) on a builder over a
27/// [`RedisStream`] or a [`RedisList`], before or after the window `.pipeline()` opens and the
28/// position `start_at(..)` names. The trait
29/// is bound to those sources, so the method does not appear on a builder for another broker or
30/// for a form that has no blocking read.
31///
32/// # Examples
33///
34/// ```
35/// # mod demo {
36/// use std::time::Duration;
37///
38/// use ruststream_fred::stream::prelude::*;
39/// # use serde::Deserialize;
40/// # #[derive(Deserialize)]
41/// # struct Order {
42/// #     amount: u64,
43/// # }
44///
45/// #[subscriber(RedisStream::new("orders").group("workers"))]
46/// async fn bill(orders: &[Order]) -> HandlerOutcome {
47///     let total: u64 = orders.iter().map(|order| order.amount).sum();
48///     tracing::info!(orders = orders.len(), total, "billed");
49///     HandlerOutcome::ack()
50/// }
51///
52/// #[ruststream::app]
53/// fn app() -> impl App {
54///     RustStream::new(AppInfo::new("billing", "0.1.0")).with_broker(
55///         RedisBroker::standalone("redis://localhost:6379"),
56///         |b| {
57///             // The size is the core's word, the block is Redis's, and they read in that order.
58///             b.include(bill.batch(nonzero!(6)).block(Duration::from_secs(5)));
59///         },
60///     )
61/// }
62/// # }
63/// ```
64pub trait RedisSubscribeExt {
65    /// How long one read blocks waiting for entries, overriding what the descriptor named.
66    ///
67    /// On a stream this is the `XREADGROUP` server-side `BLOCK` (and, in reclaim mode, the poll
68    /// interval between empty `XAUTOCLAIM` scans); on a list it is the `BRPOP` / `BLMOVE` timeout.
69    /// Both default to five seconds.
70    #[must_use]
71    fn block(self, block: Duration) -> Self;
72}
73
74impl<Def, Source, State, DefCodec> RedisSubscribeExt
75    for SubscriberBuilder<Def, Source, State, DefCodec>
76where
77    Def: Declared,
78    Source: blocking::Blocks,
79{
80    fn block(self, block: Duration) -> Self {
81        self.map_source(|source| source.block(block))
82    }
83}
84
85mod blocking {
86    use std::time::Duration;
87
88    use ruststream::StartAt;
89
90    use crate::list::RedisList;
91    use crate::pipeline::Pipelined;
92    use crate::stream::RedisStream;
93
94    /// A source with a blocking read, which is what [`block`](super::RedisSubscribeExt::block)
95    /// reaches into wherever the mount chain has wrapped it.
96    pub trait Blocks {
97        fn block(self, block: Duration) -> Self;
98    }
99
100    impl Blocks for RedisStream {
101        fn block(self, block: Duration) -> Self {
102            Self::block(self, block)
103        }
104    }
105
106    impl Blocks for RedisList {
107        fn block(self, block: Duration) -> Self {
108            Self::block(self, block)
109        }
110    }
111
112    /// The same option on a source whose window `.pipeline()` has opened, so the base settings
113    /// read in either order around it.
114    impl<Descriptor: Blocks, Mode> Blocks for Pipelined<Descriptor, Mode> {
115        fn block(self, block: Duration) -> Self {
116            self.map(|descriptor| descriptor.block(block))
117        }
118    }
119
120    /// The same option after the core's `start_at(..)`, which wraps the source it positions.
121    impl<Source: Blocks, Position> Blocks for StartAt<Source, Position> {
122        fn block(self, block: Duration) -> Self {
123            self.map_inner(|source| source.block(block))
124        }
125    }
126}
ruststream_fred/settings.rs

ruststream_fred/
settings.rs
