Prelude

One glob, ruststream_amqp::prelude, brings the core prelude, the broker, the descriptor and the policies under Publish and TransactionalPublish.

use ruststream_amqp::prelude::*; is the one import an AMQP 1.0 service writes. It carries the core's own prelude, so #[subscriber], RustStream, HandlerOutcome, Out and nonzero! come with it. It adds AmqpBroker with its authentication Sasl, the address descriptor AmqpAddress with its delivery guarantee Settle, the crate's error AmqpError, and the capability trait RequestReply.

The publish policies come under the names every broker crate uses: Publish is AmqpPublish, and with the transaction feature TransactionalPublish is AmqpTransactionalPublish, next to the capability trait TransactionalPublisher. So a mount site reads the same on every broker. Two broker preludes in one file collide on these names; the prefixed originals, exported here too, tell them apart.

A handler body bounds its slots with capability traits, such as Out<impl Publisher> or Out<impl RequestReply>, and names no broker type. A service that reads a delivery's partition key through the Partitioned capability imports ruststream::Partitioned itself. In the glob it would make partition_key() ambiguous with the method of the same name that the core prelude brings.

Example

The broker, its authentication, the descriptor with its settings and the reply policy all come from the one glob.

Macros

examples/amqp_broker_prelude.rs

use ruststream_amqp::prelude::*;

#[subscriber(
    AmqpAddress::queue("orders")
        .credit(nonzero!(64))
        .settle(Settle::AtLeastOnce),
    reply
)]
async fn confirm(order: &Order) -> Confirmation {
    Confirmation { id: order.id }
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        AmqpBroker::new("amqp://broker.internal:5672").sasl(Sasl::anonymous()),
        |b| {
            b.include(confirm).out_reply(Publish);
        },
    )
}

Manual

examples/manual/amqp_broker_prelude.rs

use ruststream_amqp::prelude::*;

struct Confirm;

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

fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        AmqpBroker::new("amqp://broker.internal:5672").sasl(Sasl::anonymous()),
        |b| {
            let confirm = subscriber(
                AmqpAddress::queue("orders")
                    .credit(nonzero!(64))
                    .settle(Settle::AtLeastOnce),
                Confirm,
            );
            b.include(confirm.reply().build()).out_reply(Publish);
        },
    )
}