ruststream_rdkafka/subscription/
topics.rs

1//! Several topics, or a topic regex, consumed as one subscription.
2
3#[cfg(feature = "asyncapi")]
4use ruststream::asyncapi::Bindings;
5use ruststream::runtime::IntoSource;
6use ruststream::{NamedCopies, SubscriptionSource};
7
8use super::{Assignment, Commit, GroupSettings, LaneKey, Reader, StartOffset, SubscriptionPlan};
9use crate::broker::ConnectedKafkaBroker;
10use crate::error::KafkaError;
11use crate::subscriber::KafkaSubscriber;
12
13/// One entry of the subscribed set: librdkafka reads a leading `^` as "this is a regex", so
14/// which of the two a name is belongs to the name, not to a flag beside the list.
15#[derive(Debug, Clone)]
16enum Subscribed {
17    Topic(String),
18    Pattern(String),
19}
20
21impl Subscribed {
22    fn as_str(&self) -> &str {
23        match self {
24            Self::Topic(name) | Self::Pattern(name) => name,
25        }
26    }
27
28    fn validate(&self) -> Result<(), KafkaError> {
29        super::reject_empty(self.as_str())?;
30        match self {
31            Self::Topic(name) => super::reject_pattern(name, "a topic of `KafkaTopics`"),
32            Self::Pattern(pattern) if pattern.starts_with('^') => Ok(()),
33            Self::Pattern(pattern) => Err(KafkaError::InvalidOptions(format!(
34                "pattern {pattern:?} must start with '^' (librdkafka's anchor for topic \
35                 regexes); without it the name would be subscribed literally"
36            ))),
37        }
38    }
39}
40
41/// A subscription to several topics, or to every topic matching a regex, through one consumer
42/// and one consumer group.
43///
44/// All matched topics share the handler, and therefore its payload type; each delivery still
45/// reports the topic it came from. Topics created after the group formed are picked up on the
46/// next metadata refresh of a pattern subscription.
47///
48/// One such subscription reads many destinations and a copy published to any one of them would
49/// come back on the wrong topic, so this form leaves the retry destination to the mount site:
50/// `.out_retry(policy).to("orders.retry")`, or a transform that names it per delivery. One
51/// topic on its own is [`KafkaTopic`](crate::KafkaTopic), which addresses its own copies.
52///
53/// # Examples
54///
55/// ```
56/// use ruststream_rdkafka::prelude::*;
57/// # #[derive(serde::Deserialize)]
58/// # struct Order { id: u64 }
59///
60/// #[subscriber(KafkaTopics::new(["orders-eu", "orders-us"]).group("orders-svc"))]
61/// async fn place(order: &Order, Ctx(topic): Ctx<Topic>) -> HandlerOutcome {
62///     tracing::info!(order.id, %topic, "accepted");
63///     HandlerOutcome::ack()
64/// }
65///
66/// fn app() -> RustStream {
67///     RustStream::new(AppInfo::new("orders", "0.1.0"))
68///         .with_broker(KafkaBroker::new(["localhost:9092"]), |b| {
69///             b.include(place);
70///         })
71/// }
72/// # let _ = app;
73/// ```
74#[derive(Debug, Clone)]
75pub struct KafkaTopics {
76    entries: Vec<Subscribed>,
77    /// The subscribed names joined with `,`: the handler metadata name, kept as an owned string
78    /// because the trait hands it out as a borrow.
79    name: String,
80    settings: GroupSettings,
81}
82
83impl KafkaTopics {
84    fn from_entries(entries: Vec<Subscribed>) -> Self {
85        let name = entries
86            .iter()
87            .map(Subscribed::as_str)
88            .collect::<Vec<_>>()
89            .join(",");
90        Self {
91            entries,
92            name,
93            settings: GroupSettings::default(),
94        }
95    }
96
97    fn push(mut self, entry: Subscribed) -> Self {
98        if !self.name.is_empty() {
99            self.name.push(',');
100        }
101        self.name.push_str(entry.as_str());
102        self.entries.push(entry);
103        self
104    }
105
106    /// Describes a subscription to every one of `topics`, by literal name.
107    #[must_use]
108    pub fn new(topics: impl IntoIterator<Item = impl Into<String>>) -> Self {
109        Self::from_entries(
110            topics
111                .into_iter()
112                .map(|topic| Subscribed::Topic(topic.into()))
113                .collect(),
114        )
115    }
116
117    /// Describes a subscription to every existing topic matching `pattern`.
118    ///
119    /// The pattern is a librdkafka topic regex and must start with `^` (that anchor is how
120    /// librdkafka distinguishes a pattern from a literal name); subscribing fails with a clear
121    /// error otherwise.
122    ///
123    /// # Examples
124    ///
125    /// ```
126    /// use ruststream_rdkafka::prelude::*;
127    /// # #[derive(serde::Deserialize)]
128    /// # struct Event { kind: String }
129    ///
130    /// #[subscriber(KafkaTopics::pattern("^audit\\..*").group("audit-svc"))]
131    /// async fn archive(event: &Event, Ctx(topic): Ctx<Topic>) -> HandlerOutcome {
132    ///     tracing::info!(kind = %event.kind, %topic, "recorded");
133    ///     HandlerOutcome::ack()
134    /// }
135    ///
136    /// fn app() -> RustStream {
137    ///     RustStream::new(AppInfo::new("audit", "0.1.0"))
138    ///         .with_broker(KafkaBroker::new(["localhost:9092"]), |b| {
139    ///             b.include(archive);
140    ///         })
141    /// }
142    /// # let _ = app;
143    /// ```
144    #[must_use]
145    pub fn pattern(pattern: impl Into<String>) -> Self {
146        Self::from_entries(vec![Subscribed::Pattern(pattern.into())])
147    }
148
149    /// Adds another literal topic to the same subscription.
150    #[must_use]
151    pub fn and_topic(self, topic: impl Into<String>) -> Self {
152        self.push(Subscribed::Topic(topic.into()))
153    }
154
155    /// Adds another `^`-anchored topic regex to the same subscription.
156    #[must_use]
157    pub fn and_pattern(self, pattern: impl Into<String>) -> Self {
158        self.push(Subscribed::Pattern(pattern.into()))
159    }
160
161    /// The consumer group for this subscription, overriding
162    /// [`KafkaBroker::default_group`](crate::KafkaBroker::default_group).
163    #[must_use]
164    pub fn group(mut self, group: impl Into<String>) -> Self {
165        self.settings.group = Some(group.into());
166        self
167    }
168
169    /// Where the group starts when it has no committed offset (see [`StartOffset`]).
170    #[must_use]
171    pub fn start(mut self, start: StartOffset) -> Self {
172        self.settings.start = start;
173        self
174    }
175
176    /// How processed deliveries are committed (see [`Commit`]).
177    #[must_use]
178    pub fn commit(mut self, commit: Commit) -> Self {
179        self.settings.commit = commit;
180        self
181    }
182
183    /// The partition assignment strategy (see [`Assignment`]); unset means the librdkafka
184    /// default (`range,roundrobin`).
185    #[must_use]
186    pub fn assignment(mut self, assignment: Assignment) -> Self {
187        self.settings.assignment = Some(assignment);
188        self
189    }
190
191    /// What drives keyed worker lanes for this subscription (see [`LaneKey`]).
192    #[must_use]
193    pub fn lane_key(mut self, lane_key: LaneKey) -> Self {
194        self.settings.lane_key = lane_key;
195        self
196    }
197
198    /// Raw librdkafka consumer property passthrough, applied last (it wins over the typed
199    /// options and the broker-wide config).
200    ///
201    /// A property the subscription's [`Commit`] mode owns is the exception; see
202    /// [`KafkaTopic::config`](crate::KafkaTopic::config).
203    #[must_use]
204    pub fn config(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
205        self.settings.config.set(key.into(), value.into());
206        self
207    }
208
209    /// The subscribed names joined with `,`, which is also the handler metadata name.
210    #[must_use]
211    pub fn topics(&self) -> &str {
212        &self.name
213    }
214
215    /// What this subscription adds to its `receive` operation: the consumer group it joins, and
216    /// the client id where the raw passthrough names one.
217    ///
218    /// There is no channel binding to go with it: the Kafka binding's `topic` names one topic,
219    /// and this subscription reads a set.
220    #[cfg(feature = "asyncapi")]
221    fn operation_bindings(&self) -> Bindings {
222        crate::bindings::operation(self.settings.group.as_deref(), &self.settings.config)
223    }
224
225    fn into_plan(self) -> Result<SubscriptionPlan, KafkaError> {
226        if self.entries.is_empty() {
227            return Err(KafkaError::InvalidOptions(
228                "`KafkaTopics` subscribes to at least one name; pass the topics the handler \
229                 consumes"
230                    .to_owned(),
231            ));
232        }
233        for entry in &self.entries {
234            entry.validate()?;
235        }
236        let name = self.name;
237        super::reject_commit_mode_clash(&name, &self.settings.commit, &self.settings.config)?;
238        let names = self
239            .entries
240            .into_iter()
241            .map(|entry| match entry {
242                Subscribed::Topic(name) | Subscribed::Pattern(name) => name,
243            })
244            .collect();
245        Ok(SubscriptionPlan {
246            name,
247            reader: Reader::Subscribed(names),
248            settings: self.settings,
249        })
250    }
251}
252
253// The manual path binds a `Handle` body to its source with `subscriber(source, body)`, which
254// takes the descriptor as it is, the way the `#[subscriber(..)]` attribute does.
255impl IntoSource for KafkaTopics {
256    type Source = Self;
257
258    fn into_source(self) -> Self {
259        self
260    }
261}
262
263impl SubscriptionSource<ConnectedKafkaBroker> for KafkaTopics {
264    type Subscriber = KafkaSubscriber;
265    type Copies = NamedCopies;
266
267    fn name(&self) -> &str {
268        &self.name
269    }
270
271    async fn subscribe(
272        self,
273        connected: &ConnectedKafkaBroker,
274    ) -> Result<Self::Subscriber, KafkaError> {
275        connected.open(self.into_plan()?).await
276    }
277
278    #[cfg(feature = "asyncapi")]
279    fn operation_bindings(&self) -> Bindings {
280        Self::operation_bindings(self)
281    }
282}
283
284#[cfg(test)]
285mod tests {
286    use super::*;
287
288    #[test]
289    fn a_list_subscribes_to_every_name() {
290        let plan = KafkaTopics::new(["orders", "cancellations"])
291            .group("orders-svc")
292            .into_plan()
293            .expect("two literal topics are a valid subscription");
294        assert_eq!(plan.name, "orders,cancellations");
295        assert!(
296            matches!(plan.reader, Reader::Subscribed(names) if names == ["orders", "cancellations"])
297        );
298    }
299
300    #[test]
301    fn an_unanchored_pattern_is_refused() {
302        let err = KafkaTopics::pattern("orders\\..*")
303            .into_plan()
304            .expect_err("an unanchored pattern would be subscribed literally");
305        let KafkaError::InvalidOptions(message) = err else {
306            panic!("a refused pattern must be an options error");
307        };
308        assert!(
309            message.contains('^'),
310            "the error must name the anchor, got: {message}"
311        );
312    }
313
314    // A regex entry passed as a literal topic is the mistake the anchor rule exists for: it
315    // would subscribe to every matching topic while the mount site believes it named one.
316    #[test]
317    fn a_regex_passed_as_a_literal_topic_is_refused() {
318        let err = KafkaTopics::new(["orders"])
319            .and_topic("^audit\\..*")
320            .into_plan()
321            .expect_err("a regex is not a literal topic name");
322        assert!(matches!(err, KafkaError::InvalidOptions(_)));
323    }
324
325    #[test]
326    fn an_empty_set_is_refused() {
327        let err = KafkaTopics::new(Vec::<String>::new())
328            .into_plan()
329            .expect_err("a subscription to nothing is not a subscription");
330        assert!(matches!(err, KafkaError::InvalidOptions(_)));
331    }
332}
ruststream_rdkafka/subscription/topics.rs

ruststream_rdkafka/subscription/
topics.rs
