Operations

A receive operation reports manual acknowledgement; a send reports the delivery mode, priority and expiration its policy fixes.

Every receive operation carries ack: true. This crate acknowledges each delivery by hand, so the handler's outcome settles it.

A slot entry and a dead-letter destination each have a send operation. Its binding carries the properties every message through the policy takes. deliveryMode is 2, persistent, unless the policy sets persistent(false), which gives 1. priority and expiration, in milliseconds, appear when the policy fixes a priority or an expiration. The document describes the declaration, so a step that adjusts one message leaves it as it is.

A reply is the reply of its receive operation. Under DirectReplyTo the reply's address has the location $message.header#/reply-to: a client reads the address from the request's reply-to header.

Example

The operation receive_inventory_check reports ack: true and a reply addressed by reply-to. The operation send_inventory_events_inventory_restock reports delivery mode 2, priority 4 and an expiration of 30000.

Macros

examples/lapin_asyncapi.rs

fn service(broker: LapinBroker) -> RustStream {
    let inventory = AppInfo::new("inventory", "1.0.0");
    RustStream::new(inventory).with_broker_labeled("rabbit", broker, |b| {
        b.include(check)
            .max_attempts(nonzero!(3u32))
            .dead_letter("inventory.dead")
            .out_reply(Publish::default().exchange("answers"))
            .transform(DirectReplyTo);
        b.include(on_event)
            .out(
                Restocks,
                Publish::default()
                    .priority(4)
                    .expiration(Duration::from_secs(30)),
            )
            .build();
    })
}

Manual

examples/manual/lapin_asyncapi.rs

fn service(broker: LapinBroker) -> RustStream {
    let inventory = AppInfo::new("inventory", "1.0.0");
    RustStream::new(inventory).with_broker_labeled("rabbit", broker, |b| {
        let check = subscriber(checks(), Check).reply().to("inventory.unrouted");
        b.include(check.build())
            .max_attempts(nonzero!(3u32))
            .dead_letter("inventory.dead")
            .out_reply(Publish::default().exchange("answers"))
            .transform(DirectReplyTo);
        b.include(subscriber(events(), OnEvent).build())
            .out(
                Restocks,
                Publish::default()
                    .priority(4)
                    .expiration(Duration::from_secs(30)),
            )
            .build();
    })
}