Struct ToSourceTopic 

pub struct ToSourceTopic;

A PublishTransform sending each retry copy back to the topic its delivery arrived on.

A KafkaTopics subscription reads several topics, and a KafkaPartitions reader reads part of one, so neither addresses its own retry copies and every registration over them names where a copy goes. Naming one fixed topic with .to(topic) sends an orders-eu record’s retry to that topic whatever it was; this transform reads the record’s own topic out of the delivery instead, so a retry stays where it belongs and the handler sees it arrive under the same name.

It writes no per-record setting, so it mounts over any Kafka publisher, and it takes the destination, so it mounts where nothing else has declared one - a retry position with no .to(topic).

The topic comes from KafkaContext, which reaches the position when the handler reads that context (any Ctx<..> key of context::keys). A batch handler carries KafkaBatchContext instead, which holds no per-record topic, so a batch registration names its destination with .to(topic).

Examples

use ruststream_rdkafka::prelude::*;

/// Reading the source topic is also what puts the Kafka context on this registration, which
/// is what the transform reads.
#[subscriber(KafkaTopics::new(["orders-eu", "orders-us"]).group("orders-svc"))]
async fn place(order: &Order, Ctx(topic): Ctx<Topic>) -> HandlerOutcome {
    tracing::info!(order.id, %topic, "accepted");
    HandlerOutcome::ack()
}

fn app() -> RustStream {
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .with_broker(KafkaBroker::new(["localhost:9092"]), |b| {
            b.include(place)
                .max_attempts(nonzero!(5u32))
                .dead_letter("orders.dlq")
                .out_retry(Publish::default())
                .transform(ToSourceTopic);
        })
}

Trait Implementations

impl Clone for ToSourceTopic

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Copy for ToSourceTopic

impl Debug for ToSourceTopic

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl Default for ToSourceTopic

fn default() -> Self

Returns the “default value” for a type.

impl Eq for ToSourceTopic

impl PartialEq for ToSourceTopic

fn eq(&self, other: &Self) -> bool

Equality operator ==.

fn ne(&self, other: &Rhs) -> bool

Inequality operator !=.

impl<Options> PublishTransform<ForReply<KafkaContext>, Options> for ToSourceTopic

type Destination = Names

What this transform does to the destination: Reads or Names.

fn apply( &self, out: &mut Outgoing<'_>, _options: &mut Option<Options>, cx: &PublishContext<'_, KafkaContext>, )

Transforms out in place before it is sent, reading the position’s view through cx and resolving the broker’s per-message settings through options.

impl StructuralPartialEq for ToSourceTopic