Struct RedisPublishOptions 

pub struct RedisPublishOptions {
    pub partition_key: Option<Vec<u8>>,
    pub join_round: bool,
}

The per-message settings of every publisher and transaction in this crate.

One field, the partition key, and it is optional like every field of an options type: a publish that names no step carries no options at all, and nothing is written into its headers.

The type is the bound that keeps this crate’s builder steps off another broker’s publisher, and it is what a handler body names when it adjusts a setting (Out<impl Publisher<Options = RedisPublishOptions>, Marker>). A test reads the value back off the slot view with with_options.

Examples

use std::error::Error;

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

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

#[derive(OutSlot)]
#[publishes(Order)]
struct Ledger;

#[subscriber(RedisStream::new("orders.in").group("workers"))]
async fn forward(
    order: &Order,
    Out(ledger): Out<impl Publisher<Options = RedisPublishOptions>, Ledger>,
) -> HandlerOutcome {
    let sent = ledger
        .message(order)
        .to("orders.keyed")
        .partition_key(&order.tenant)
        .publish()
        .await;
    match sent {
        Ok(_) => HandlerOutcome::ack(),
        Err(_) => HandlerOutcome::retry(),
    }
}

pub fn app() -> impl App<State = ()> {
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .with_broker(RedisBroker::standalone("redis://localhost:6379"), |b| {
            b.include(forward).out(Ledger, Publish).build();
        })
}

pub async fn the_ledger_is_keyed_by_tenant() -> Result<(), Box<dyn Error>> {
    let tb = TestApp::start(app()).await?;
    let order = Order {
        tenant: "tenant-a".to_owned(),
        id: 7,
    };
    tb.broker::<RedisBroker>()
        .message(&order)
        .to("orders.in")
        .publish()
        .await?;

    tb.out::<Ledger>()
        .assert_called_once()
        .with_options(&RedisPublishOptions {
            partition_key: Some(b"tenant-a".to_vec()),
            ..RedisPublishOptions::default()
        });
    Ok(())
}

Fields

partition_key: Option<Vec<u8>>

The key this one message is partitioned by, or None to send it unkeyed.

Opaque bytes: the runtime hashes them to pick a dispatch lane, and never interprets them.

join_round: bool

Whether this message joins the round of the delivery being handled, on a .pipeline() subscription, instead of leaving at once.

The InRound transform sets it on a reply; a message sent outside a pipelined delivery’s handling leaves at once whatever it says.

Trait Implementations

impl Clone for RedisPublishOptions

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 RedisPublishOptions

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

Formats the value using the given formatter.

impl Default for RedisPublishOptions

fn default() -> Self

Returns the “default value” for a type.

impl Eq for RedisPublishOptions

impl PartialEq for RedisPublishOptions

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

Equality operator ==.

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

Inequality operator !=.

impl<C> PublishTransform<ForReply<C>, RedisPublishOptions> for InRound

type Destination = Reads

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

fn apply( &self, _out: &mut Outgoing<'_>, options: &mut Option<RedisPublishOptions>, _cx: &PublishContext<'_, C>, )

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 RedisPublishOptions