Struct AmqpAddress 

pub struct AmqpAddress { /* private fields */ }

A subscription descriptor for an AMQP 1.0 address.

Implements SubscriptionSource, so it can sit inline in the #[subscriber(..)] decorator:

use ruststream_amqp::prelude::*;
use serde::Deserialize;

#[derive(Deserialize)]
struct Order {
    id: u64,
}

#[subscriber(AmqpAddress::queue("orders").credit(nonzero!(64)))]
async fn fulfil(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "fulfilling");
    HandlerOutcome::ack()
}

Implementations

impl AmqpAddress

pub fn queue(name: impl Into<String>) -> Self

An anycast address: competing consumers, each message delivered to one of them.

Examples

use ruststream_amqp::prelude::*;
use serde::Deserialize;

#[derive(Deserialize)]
struct Order {
    id: u64,
}

// Every replica of the service competes for "orders": each order is handled once.
#[subscriber(AmqpAddress::queue("orders"))]
async fn fulfil(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "fulfilling");
    HandlerOutcome::ack()
}

pub fn topic(name: impl Into<String>) -> Self

A multicast address: fan-out, each message delivered to every subscriber.

Examples

use ruststream_amqp::prelude::*;
use serde::Deserialize;

#[derive(Deserialize)]
struct Event {
    kind: String,
}

// Every service subscribed to "events" gets its own copy of each event.
#[subscriber(AmqpAddress::topic("events"))]
async fn audit(event: &Event) -> HandlerOutcome {
    tracing::info!(kind = %event.kind, "audited");
    HandlerOutcome::ack()
}

pub fn raw(address: impl Into<String>) -> Self

A verbatim address, for deployments with their own addressing convention ("/queues/orders" on RabbitMQ 4.x, a fully qualified queue on Artemis).

Examples

use ruststream_amqp::prelude::*;
use serde::Deserialize;

#[derive(Deserialize)]
struct Order {
    id: u64,
}

// RabbitMQ 4.x addresses a queue by its v2 path; the string reaches the broker as written.
#[subscriber(AmqpAddress::raw("/queues/orders"))]
async fn fulfil(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "fulfilling");
    HandlerOutcome::ack()
}

pub fn credit(self, credit: NonZeroU32) -> Self

Sets the link credit: the window of deliveries the broker may send ahead of the handler’s settlements. Defaults to DEFAULT_CREDIT.

The subscription grants the full credit when it attaches, and grants it again each time half of it has been settled. The broker then never runs more than credit deliveries ahead of the last grant, but the deliveries held unsettled can outnumber the credit: with a credit of 4, the second settlement brings four more while two are still open.

The count is a NonZeroU32 because a subscription granted no credit receives nothing: zero is not a quieter setting but a stalled subscription, so it is unrepresentable here.

Examples

use ruststream_amqp::prelude::*;
use serde::Deserialize;

#[derive(Deserialize)]
struct Order {
    id: u64,
}

// The broker sends up to 64 orders ahead of the handler's settlements.
#[subscriber(AmqpAddress::queue("orders").credit(nonzero!(64)))]
async fn fulfil(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "fulfilling");
    HandlerOutcome::ack()
}

A zero is rejected while the service is compiled, not when the subscription opens:

use ruststream::nonzero;
use ruststream_amqp::AmqpAddress;

let source = AmqpAddress::queue("orders").credit(nonzero!(0));

pub fn settle(self, settle: Settle) -> Self

Sets the delivery guarantee. Defaults to Settle::AtLeastOnce.

Examples

use ruststream_amqp::prelude::*;
use serde::Deserialize;

#[derive(Deserialize)]
struct Event {
    kind: String,
}

// Settled on receipt: the audit may miss an event when the process stops, and the broker
// never waits for it.
#[subscriber(AmqpAddress::topic("events").settle(Settle::AtMostOnce))]
async fn audit(events: &[Event]) -> HandlerOutcome {
    for event in events {
        tracing::info!(kind = %event.kind, "audited");
    }
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("audit", "0.1.0"))
        .with_broker(AmqpBroker::new("amqp://localhost:5672"), |b| {
            b.include(audit.batch(nonzero!(32)));
        })
}

pub fn batch_wait(self, batch_wait: Duration) -> Self

Caps how long a partial batch waits for more deliveries on a batch subscription, counted from its first one. Defaults to DEFAULT_BATCH_WAIT.

AMQP 1.0 has no batch pull, so the batches are assembled on the client and this deadline is what trades latency for fuller batches under a trickle of traffic. It has no effect on a single-message subscription, where every delivery goes out as it arrives.

Examples

use std::time::Duration;

use ruststream_amqp::prelude::*;
use serde::Deserialize;

#[derive(Deserialize)]
struct Order {
    id: u64,
}

// Up to 32 orders per call, or whatever arrived within 50 ms of the first one.
#[subscriber(AmqpAddress::queue("orders").batch_wait(Duration::from_millis(50)))]
async fn fulfil(orders: &[Order]) -> HandlerOutcome {
    tracing::info!(count = orders.len(), "fulfilling");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .with_broker(AmqpBroker::new("amqp://localhost:5672"), |b| {
            b.include(fulfil.batch(nonzero!(32)));
        })
}

pub fn address(&self) -> &str

The address string sent to the broker.

Trait Implementations

impl Clone for AmqpAddress

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for AmqpAddress

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl Eq for AmqpAddress

impl IntoSource for AmqpAddress

The descriptor mounts on the manual path the way it does on the attribute: the subscriber(..) constructor takes it as its source, unchanged.

Examples

use std::future::{Future, ready};

use ruststream_amqp::prelude::*;
use serde::Deserialize;

#[derive(Deserialize)]
struct Order {
    id: u64,
}

struct TakeOrder;

impl Handle<Order> for TakeOrder {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(order.id, "order taken");
        ready(Ok(()))
    }
}

fn app() -> RustStream {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        AmqpBroker::new("amqp://localhost:5672"),
        |b| {
            b.include(subscriber(AmqpAddress::queue("orders"), TakeOrder).build());
        },
    )
}

type Source = AmqpAddress

The source the constructor stores.

fn into_source(self) -> Self

Builds the source.

impl PartialEq for AmqpAddress

fn eq(&self, other: &Self) -> bool

Equality operator ==.

fn ne(&self, other: &Rhs) -> bool

Inequality operator !=.

impl RedeliveryAddressed<ConnectedAmqpBroker> for AmqpAddress

fn redelivery_address( &self, _connected: &ConnectedAmqpBroker, ) -> impl Future<Output = Result<RedeliveryAddress, AmqpError>> + Send

Where a publish reaches this subscription again.

impl StructuralPartialEq for AmqpAddress

impl SubscriptionSource<ConnectedAmqpBroker> for AmqpAddress

type Copies = AddressedCopies

AMQP 1.0 has no delayed redelivery and no wire-level way to declare a node with a delivery limit and a dead-letter address, so a spent delivery is this process’s to move: the runtime publishes the copies and this descriptor says where they land.

type Subscriber = AmqpSubscriber

The subscriber type this source opens.

fn name(&self) -> &str

The name (subject / channel) this subscription binds to.

async fn subscribe( self, connected: &ConnectedAmqpBroker, ) -> Result<AmqpSubscriber, AmqpError>

Opens the subscription against the connected broker. Called once at startup.

fn channel_bindings(&self) -> Bindings

What this subscription adds to its channel in the generated AsyncAPI document.

fn declare_retry(self, declaration: &RetryDeclaration) -> Self

where Self: Sized,

Takes the registration’s retry declaration into the descriptor, before the subscription opens.

fn declare_retry_on( &self, connected: &C, declaration: &RetryDeclaration, ) -> Result<(), DeclareRetryError>

Carries the registration’s declaration to the broker, at the same point and before the subscription opens.

fn operation_bindings(&self) -> Bindings

What this subscription adds to its receive operation in the document.

fn message_bindings(&self) -> Bindings

What this subscription adds to the messages that arrive on it.