Options

A broker's per-message settings, such as a priority, a QoS or a key: defaulted by the publish policy, set per publish by the broker's own steps.

A broker declares its settings as Publisher::Options, a type of its own with every field optional. The publish policy at the mount site holds the defaults. A body that sets one bounds its slot Out<impl Publisher<Options = ..>, _> and imports the broker's prelude, whose extension trait adds the steps to the builder. A step sets one field for one publish; a field no step touches keeps the policy's default. A reply and a retry copy have no call site, so a transform that writes the broker's options sets them there. A transform runs after the call site, so it wins a field both of them set. A setting reaches the broker's publisher as a value, not as a header, and a test reads it back with with_options.

The memory broker's publisher has no settings, so the example wraps it with a priority the way a networked broker ships one, and turns the priority into a header the test can read. A service on a networked broker imports the broker's own options and steps instead.

Example

The body sets the priority of an express order; the mount site sets the default.

Macros

examples/per_message_options.rs

/// The bound names the broker's settings, so the broker's own steps appear on the builder.
#[subscriber("orders")]
async fn ship(
    order: &Order,
    Out(out): Out<impl Publisher<Options = PriorityOptions>, Shipments>,
) -> HandlerOutcome {
    let shipment = out.message(order).to("shipments");
    let sent = if order.express {
        shipment.priority(9).publish().await
    } else {
        shipment.publish().await
    };
    if sent.is_err() {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

Manual

examples/manual/per_message_options.rs

/// The bound names the broker's settings, so the broker's own steps appear on the builder.
struct Ship;

impl<Entry> Handle<Order, (), Outs<(Entry,)>> for Ship
where
    Entry: OutEntry<Shipments, Wire: Publisher<Options = PriorityOptions>>,
{
    async fn handle(
        &self,
        order: &Order,
        outs: &Outs<(Entry,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        let shipment = outs.get(Shipments).message(order).to("shipments");
        let sent = if order.express {
            shipment.priority(9).publish().await
        } else {
            shipment.publish().await
        };
        if sent.is_err() {
            return Err(HandlerOutcome::retry());
        }
        Ok(())
    }
}

A reply takes its settings from a transform.

Macros

examples/per_message_options.rs

/// A reply has no call site, so a transform sets its settings.
struct LowPriority;

impl<K: ContextKind> PublishTransform<K, PriorityOptions> for LowPriority {
    type Destination = Reads;

    fn apply(
        &self,
        _out: &mut Outgoing<'_>,
        options: &mut Option<PriorityOptions>,
        _cx: &K::View<'_>,
    ) {
        options
            .get_or_insert_with(PriorityOptions::default)
            .priority = Some(1);
    }
}

#[subscriber("receipt-requests", reply)]
async fn issue_receipt(order: &Order) -> Receipt {
    Receipt { order_id: order.id }
}

Manual

examples/manual/per_message_options.rs

/// A reply has no call site, so a transform sets its settings.
struct LowPriority;

impl<K: ContextKind> PublishTransform<K, PriorityOptions> for LowPriority {
    type Destination = Reads;

    fn apply(
        &self,
        _out: &mut Outgoing<'_>,
        options: &mut Option<PriorityOptions>,
        _cx: &K::View<'_>,
    ) {
        options
            .get_or_insert_with(PriorityOptions::default)
            .priority = Some(1);
    }
}

struct IssueReceipt;

impl Handle<Order, Receipt> for IssueReceipt {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Receipt, HandlerOutcome>> {
        ready(Ok(Receipt { order_id: order.id }))
    }
}

Macros

examples/per_message_options.rs

fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        // The policy holds the defaults: priority 3 unless a step says otherwise.
        b.include(ship)
            .out(Shipments, PriorityPublish::default().priority(3))
            .build();
        b.include(issue_receipt)
            .out_reply(PriorityPublish::default().priority(3))
            .transform(LowPriority);
    })
}

Manual

examples/manual/per_message_options.rs

fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        // The policy holds the defaults: priority 3 unless a step says otherwise.
        b.include(subscriber("orders", Ship).build())
            .out(Shipments, PriorityPublish::default().priority(3))
            .build();
        b.include(subscriber("receipt-requests", IssueReceipt).reply().build())
            .out_reply(PriorityPublish::default().priority(3))
            .transform(LowPriority);
    })
}