Properties

LapinPublishSteps adds priority, expiration and persistent to the publish builder, so a handler sets an AMQP property of one message.

A publish policy holds the defaults for every message it sends, and a step adjusts one message over them. LapinPublishOptions holds the three settings, each optional: what a call leaves alone keeps the policy's value. The steps exist on a publisher whose options type is LapinPublishOptions. So the handler bounds its slot Out<impl Publisher<Options = LapinPublishOptions>, Slot> and still names no publisher type. The steps come from this crate's prelude, the one prelude a handler body imports besides the framework's. The values reach the frame as protocol fields, never as headers. The publish still leaves through the slot's own entry, with its codec and transforms. A test reads the options back with with_options on tb.out::<Slot>().

Example

An expedited order leaves at priority 9 and expires after an hour. An ordinary order takes the policy's priority 3.

Macros

examples/lapin_publishing_properties.rs

#[derive(OutSlot)]
#[publishes(Shipment)]
struct Shipments;

/// An expedited order leaves at priority 9 and expires after an hour. An ordinary one takes the
/// policy's defaults.
#[subscriber(RabbitQueue::new("orders"))]
async fn ship(
    order: &Order,
    Out(shipments): Out<impl Publisher<Options = LapinPublishOptions>, Shipments>,
) -> HandlerOutcome {
    let shipment = Shipment { order_id: order.id };
    let sent = if order.expedited {
        shipments
            .message(&shipment)
            .priority(9)
            .expiration(Duration::from_hours(1))
            .publish()
            .await
    } else {
        shipments.message(&shipment).publish().await
    };
    if sent.is_err() {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        LapinBroker::new("amqp://localhost:5672"),
        |b| {
            b.include(ship)
                .out(Shipments, Publish::default().priority(3))
                .build();
        },
    )
}

Manual

examples/manual/lapin_publishing_properties.rs

#[derive(OutSlot)]
#[publishes(Shipment)]
struct Shipments;

/// An expedited order leaves at priority 9 and expires after an hour. An ordinary one takes the
/// policy's defaults.
struct Ship;

impl<Egress> Handle<Order, (), Outs<(Egress,)>> for Ship
where
    Egress: OutEntry<Shipments, Wire: Publisher<Options = LapinPublishOptions>>,
{
    async fn handle(
        &self,
        order: &Order,
        outs: &Outs<(Egress,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        let shipment = Shipment { order_id: order.id };
        let shipments = outs.get(Shipments);
        let sent = if order.expedited {
            shipments
                .message(&shipment)
                .priority(9)
                .expiration(Duration::from_hours(1))
                .publish()
                .await
        } else {
            shipments.message(&shipment).publish().await
        };
        if sent.is_err() {
            return Err(HandlerOutcome::retry());
        }
        Ok(())
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        LapinBroker::new("amqp://localhost:5672"),
        |b| {
            b.include(subscriber(RabbitQueue::new("orders"), Ship).build())
                .out(Shipments, Publish::default().priority(3))
                .build();
        },
    )
}