Struct KafkaOptions 

pub struct KafkaOptions { /* private fields */ }

What one Kafka record carries beyond its topic, payload and headers.

A publish builder writes these through the steps of KafkaPublishSteps; nothing else needs to name the type, except an assertion reading back what a publish carried (tb.out::<Marker>().with_options(&KafkaOptions::default().partition(3))).

The record key is deliberately absent: it is the core’s Partitioned contract, carried by the PARTITION_KEY_HEADER header so that broker-agnostic code can set it, and Kafka maps it onto the native key.

Examples

A test reading back the partition a handler’s publish asked for:

use std::error::Error;

use ruststream::testing::TestApp;
use ruststream_rdkafka::prelude::*;
use serde::{Deserialize, Serialize};

#[derive(Serialize, Deserialize, Outgoing)]
struct Order {
    id: u64,
}

#[derive(Serialize, Outgoing)]
#[outgoing(name = "audit")]
struct Audit {
    id: u64,
}

#[derive(OutSlot)]
#[publishes(Audit)]
struct Journal;

#[subscriber("orders")]
async fn record(
    order: &Order,
    Out(journal): Out<impl Publisher<Options = KafkaOptions>, Journal>,
) -> HandlerOutcome {
    let sent = journal.message(&Audit { id: order.id }).partition(3).publish().await;
    if sent.is_err() {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

pub fn app() -> RustStream {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        KafkaBroker::new(["kafka:9092"]).default_group("orders-svc"),
        |b| {
            b.include(record).out(Journal, Publish::default()).build();
        },
    )
}

pub async fn audits_go_to_partition_three() -> Result<(), Box<dyn Error + Send + Sync>> {
    let tb = TestApp::start(app()).await?;

    tb.broker::<KafkaBroker>()
        .message(&Order { id: 7 })
        .to("orders")
        .publish()
        .await?;

    tb.out::<Journal>()
        .assert_called(1)
        .with_options(&KafkaOptions::default().partition(3));
    tb.shutdown().await?;
    Ok(())
}

Implementations

impl KafkaOptions

pub const fn partition(self, partition: i32) -> Self

Produces the record to exactly this partition, ahead of the partitioner and the record key.

pub const fn partition_setting(self) -> Option<i32>

The partition this record is pinned to, or None when the partitioner places it.

Trait Implementations

impl Clone for KafkaOptions

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 KafkaOptions

impl Debug for KafkaOptions

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

Formats the value using the given formatter.

impl Default for KafkaOptions

fn default() -> Self

Returns the “default value” for a type.

impl Eq for KafkaOptions

impl PartialEq for KafkaOptions

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

Equality operator ==.

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

Inequality operator !=.

impl<K: ContextKind> PublishTransform<K, KafkaOptions> for RoundRobin

type Destination = Reads

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

fn apply( &self, out: &mut Outgoing<'_>, options: &mut Option<KafkaOptions>, _cx: &K::View<'_>, )

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 KafkaOptions