Struct KafkaTopics 

pub struct KafkaTopics { /* private fields */ }

A subscription to several topics, or to every topic matching a regex, through one consumer and one consumer group.

All matched topics share the handler, and therefore its payload type; each delivery still reports the topic it came from. Topics created after the group formed are picked up on the next metadata refresh of a pattern subscription.

One such subscription reads many destinations and a copy published to any one of them would come back on the wrong topic, so this form leaves the retry destination to the mount site: .out_retry(policy).to("orders.retry"), or a transform that names it per delivery. One topic on its own is KafkaTopic, which addresses its own copies.

Examples

use ruststream_rdkafka::prelude::*;

#[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);
        })
}

Implementations

impl KafkaTopics

pub fn new(topics: impl IntoIterator<Item = impl Into<String>>) -> Self

Describes a subscription to every one of topics, by literal name.

pub fn pattern(pattern: impl Into<String>) -> Self

Describes a subscription to every existing topic matching pattern.

The pattern is a librdkafka topic regex and must start with ^ (that anchor is how librdkafka distinguishes a pattern from a literal name); subscribing fails with a clear error otherwise.

Examples

use ruststream_rdkafka::prelude::*;

#[subscriber(KafkaTopics::pattern("^audit\\..*").group("audit-svc"))]
async fn archive(event: &Event, Ctx(topic): Ctx<Topic>) -> HandlerOutcome {
    tracing::info!(kind = %event.kind, %topic, "recorded");
    HandlerOutcome::ack()
}

fn app() -> RustStream {
    RustStream::new(AppInfo::new("audit", "0.1.0"))
        .with_broker(KafkaBroker::new(["localhost:9092"]), |b| {
            b.include(archive);
        })
}

pub fn and_topic(self, topic: impl Into<String>) -> Self

Adds another literal topic to the same subscription.

pub fn and_pattern(self, pattern: impl Into<String>) -> Self

Adds another ^-anchored topic regex to the same subscription.

pub fn group(self, group: impl Into<String>) -> Self

The consumer group for this subscription, overriding KafkaBroker::default_group.

pub fn start(self, start: StartOffset) -> Self

Where the group starts when it has no committed offset (see StartOffset).

pub fn commit(self, commit: Commit) -> Self

How processed deliveries are committed (see Commit).

pub fn assignment(self, assignment: Assignment) -> Self

The partition assignment strategy (see Assignment); unset means the librdkafka default (range,roundrobin).

pub fn lane_key(self, lane_key: LaneKey) -> Self

What drives keyed worker lanes for this subscription (see LaneKey).

pub fn config(self, key: impl Into<String>, value: impl Into<String>) -> Self

Raw librdkafka consumer property passthrough, applied last (it wins over the typed options and the broker-wide config).

A property the subscription’s Commit mode owns is the exception; see KafkaTopic::config.

pub fn topics(&self) -> &str

The subscribed names joined with ,, which is also the handler metadata name.

Trait Implementations

impl Clone for KafkaTopics

fn clone(&self) -> Self

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl Debug for KafkaTopics

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

Formats the value using the given formatter.

impl IntoSource for KafkaTopics

type Source = KafkaTopics

The source the constructor stores.

fn into_source(self) -> Self

Builds the source.

impl SubscriptionSource<ConnectedKafkaBroker> for KafkaTopics

type Subscriber = KafkaSubscriber

The subscriber type this source opens.

type Copies = NamedCopies

Who publishes the copies this subscription’s retries are made of, and who names where they go: AddressedCopies, NamedCopies or BrokerMoves.

fn name(&self) -> &str

The name (subject / channel) this subscription binds to.

async fn subscribe( self, connected: &ConnectedKafkaBroker, ) -> Result<Self::Subscriber, KafkaError>

Opens the subscription against the connected broker. Called once at startup.

fn operation_bindings(&self) -> Bindings

What this subscription adds to its receive operation in the document.

fn declare_retry(self, declaration: &RetryDeclaration) -> Self

where Self: Sized,

Takes the registration’s retry declaration into the descriptor, before the subscription opens.

fn declare_retry_on( &self, connected: &C, declaration: &RetryDeclaration, ) -> Result<(), DeclareRetryError>

Carries the registration’s declaration to the broker, at the same point and before the subscription opens.

fn channel_bindings(&self) -> Bindings

What this subscription adds to its channel in the generated AsyncAPI document.

fn message_bindings(&self) -> Bindings

What this subscription adds to the messages that arrive on it.