ruststream_rdkafka/
redelivery.rs

1//! Where a retry copy goes when the subscription reads more than one topic.
2//!
3//! A subscription over a set of topics addresses no copy of its own, so the mount site names the
4//! destination. [`ToSourceTopic`] is the answer that keeps a retry on the topic it belongs to;
5//! `.to(topic)` is the one for a fixed retry topic.
6
7use ruststream::runtime::{ForReply, Names, Outgoing, PublishContext, PublishTransform};
8
9use crate::context::KafkaContext;
10use crate::context::keys::Topic;
11
12/// A [`PublishTransform`] sending each retry copy back to the topic its delivery arrived on.
13///
14/// A [`KafkaTopics`](crate::KafkaTopics) subscription reads several topics, and a
15/// [`KafkaPartitions`](crate::KafkaPartitions) reader reads part of one, so neither addresses its
16/// own retry copies and every registration over them names where a copy goes. Naming one fixed
17/// topic with `.to(topic)` sends an `orders-eu` record's retry to that topic whatever it was;
18/// this transform reads the record's own topic out of the delivery instead, so a retry stays
19/// where it belongs and the handler sees it arrive under the same name.
20///
21/// It writes no per-record setting, so it mounts over any Kafka publisher, and it takes the
22/// destination, so it mounts where nothing else has declared one - a retry position with no
23/// `.to(topic)`.
24///
25/// The topic comes from [`KafkaContext`], which reaches the position when the handler reads that
26/// context (any `Ctx<..>` key of [`context::keys`](crate::context)). A batch handler carries
27/// [`KafkaBatchContext`](crate::context::KafkaBatchContext) instead, which holds no per-record
28/// topic, so a batch registration names its destination with `.to(topic)`.
29///
30/// # Examples
31///
32/// ```
33/// # #[cfg(feature = "json")]
34/// # mod demo {
35/// use ruststream_rdkafka::prelude::*;
36/// # #[derive(serde::Deserialize)]
37/// # struct Order { id: u64 }
38///
39/// /// Reading the source topic is also what puts the Kafka context on this registration, which
40/// /// is what the transform reads.
41/// #[subscriber(KafkaTopics::new(["orders-eu", "orders-us"]).group("orders-svc"))]
42/// async fn place(order: &Order, Ctx(topic): Ctx<Topic>) -> HandlerOutcome {
43///     tracing::info!(order.id, %topic, "accepted");
44///     HandlerOutcome::ack()
45/// }
46///
47/// fn app() -> RustStream {
48///     RustStream::new(AppInfo::new("orders", "0.1.0"))
49///         .with_broker(KafkaBroker::new(["localhost:9092"]), |b| {
50///             b.include(place)
51///                 .max_attempts(nonzero!(5u32))
52///                 .dead_letter("orders.dlq")
53///                 .out_retry(Publish::default())
54///                 .transform(ToSourceTopic);
55///         })
56/// }
57/// # }
58/// ```
59#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
60pub struct ToSourceTopic;
61
62impl<Options> PublishTransform<ForReply<KafkaContext>, Options> for ToSourceTopic {
63    // Naming the destination is the whole point, so it mounts only where nothing else has.
64    type Destination = Names;
65
66    fn apply(
67        &self,
68        out: &mut Outgoing<'_>,
69        _options: &mut Option<Options>,
70        cx: &PublishContext<'_, KafkaContext>,
71    ) {
72        out.set_name(cx.context(Topic).to_owned());
73    }
74}
ruststream_rdkafka/redelivery.rs

ruststream_rdkafka/
redelivery.rs
