Prelude

One glob, ruststream_lapin::prelude, brings the core prelude, the descriptors, the policies under the family's names and the context keys.

use ruststream_lapin::prelude::*; is the one import a RabbitMQ routes file writes. It carries the core's own prelude, so #[subscriber], RustStream, HandlerOutcome and Out come with it. It adds LapinBroker, the descriptors RabbitQueue and RabbitQuorumQueue, RabbitExchange, Delay and DirectReplyTo. It adds the context keys Exchange, RoutingKey, Redelivered and DeliveryTag, and the argument types AMQPValue and FieldTable. It also carries the transaction traits a handler settles a transaction with.

The publish policies come under the names every broker crate uses: Publish is LapinPublish, TransactionalPublish is ConfirmsPublish, and Request is LapinRequest. So a mount site reads the same on every broker. ServerTxPublish keeps its own name: an AMQP server transaction is a guarantee of its own, not a second spelling of TransactionalPublish. Two broker preludes in one file collide on the uniform names; the prefixed types at the crate root tell them apart.

A handler body bounds its slots with capability traits, such as Out<impl TransactionalPublisher>, and names no broker type. The one exception is a body that sets an AMQP property for a single message: it bounds its slot Out<impl Publisher<Options = LapinPublishOptions>, Marker> and calls the steps of LapinPublishSteps, both from this prelude.

Example

The routes file reads the routing key through a context key and replies through TransactionalPublish, all from the one glob.

Macros
Manual

examples/lapin_broker_prelude.rs
use ruststream_lapin::prelude::*;

#[subscriber(
    RabbitQueue::new("orders").bind(RabbitExchange::topic("events"), "order.*"),
    reply
)]
async fn confirm(order: &Order, Ctx(routing_key): Ctx<RoutingKey>) -> Confirmation {
    tracing::info!(order.id, %routing_key, "order confirmed");
    Confirmation { id: order.id }
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        LapinBroker::new("amqp://localhost:5672"),
        |b| {
            b.include(confirm)
                .out_reply(TransactionalPublish::default());
        },
    )
}

examples/manual/lapin_broker_prelude.rs
use ruststream_lapin::prelude::*;

struct Confirm;

impl Handle<Order, Confirmation, (), AmqpContext> for Confirm {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        ctx: &mut Context<'_, AmqpContext>,
    ) -> impl Future<Output = Result<Confirmation, HandlerOutcome>> {
        let routing_key = ctx.context(RoutingKey);
        tracing::info!(order.id, %routing_key, "order confirmed");
        ready(Ok(Confirmation { id: order.id }))
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        LapinBroker::new("amqp://localhost:5672"),
        |b| {
            let orders =
                RabbitQueue::new("orders").bind(RabbitExchange::topic("events"), "order.*");
            b.include(subscriber(orders, Confirm).reply().build())
                .out_reply(TransactionalPublish::default());
        },
    )
}

Example​

Example