Struct LapinBroker 

pub struct LapinBroker { /* private fields */ }

A RabbitMQ broker backed by lapin: configuration captured, no I/O performed yet.

new is synchronous and records only the connection settings, so a RabbitMQ service is assembled with the synchronous #[ruststream::app] builder like any other broker. The runtime calls Broker::connect once at startup, which consumes this value and yields the ConnectedLapinBroker witness: subscriptions, publishers, and requesters exist only from there, so “not connected” is not representable.

By default the broker never creates infrastructure: descriptors describe the EXPECTED topology, and a missing queue is a subscribe error. Opt into declaration with declare_topology(true).

Examples

use ruststream_lapin::prelude::*;
use serde::Deserialize;

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

#[subscriber("orders")]
async fn handle(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "accepted");
    HandlerOutcome::ack()
}

fn app() -> RustStream {
    let broker = LapinBroker::new("amqp://localhost:5672")
        .prefetch(nonzero!(64))
        .declare_topology(true);
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        b.include(handle);
    })
}

Testing

With the testing feature, TestApp::start runs the service’s own app with the broker in process, and a test addresses it by its production type:

use std::error::Error;

use ruststream::testing::TestApp;
use ruststream_lapin::prelude::*;
use serde::{Deserialize, Serialize};

#[derive(Debug, Serialize, Deserialize, PartialEq, Outgoing)]
#[outgoing(name = "payments")]
struct Payment {
    amount: u64,
}

#[subscriber(RabbitQueue::new("payments"))]
async fn accept(payment: &Payment) -> HandlerOutcome {
    if payment.amount == 0 {
        return HandlerOutcome::drop();
    }
    HandlerOutcome::ack()
}

/// The app `main` runs.
pub fn app() -> RustStream {
    RustStream::new(AppInfo::new("payments", "0.1.0"))
        .with_broker(LapinBroker::new("amqp://rabbit:5672"), |b| {
            b.include(accept);
        })
}

pub async fn accepts_a_payment() -> Result<(), Box<dyn Error>> {
    let tb = TestApp::start(app()).await?;

    // The publish returns once the handler it woke has settled.
    tb.broker::<LapinBroker>()
        .message(&Payment { amount: 100 })
        .publish()
        .await?;

    tb.broker::<LapinBroker>()
        .subscriber("payments")
        .assert_called_once()
        .with(&Payment { amount: 100 })
        .settled(HandlerOutcome::ack());

    tb.shutdown().await?;
    Ok(())
}

Implementations

impl LapinBroker

pub fn new(uri: impl Into<String>) -> Self

Records the connection URI; no I/O happens until Broker::connect.

The URI carries credentials, virtual host, and TLS scheme: amqp://user:pass@host:5672/vhost (or amqps:// with a TLS feature enabled).

pub fn connection_name(self, name: impl Into<String>) -> Self

A connection name shown in the RabbitMQ management UI.

pub fn prefetch(self, prefetch: NonZeroU16) -> Self

Caps unacknowledged deliveries in flight per subscription (basic.qos).

This is the back-pressure window for subscriber streams; individual queue descriptors can override it. Without it the server imposes no prefetch limit.

The count is a NonZeroU16 because AMQP reads basic.qos(0) as “no limit”, the exact opposite of a cap: the zero sentinel is unrepresentable here, and leaving the prefetch unset is how “unlimited” is spelled.

pub fn declare_topology(self, declare: bool) -> Self

Whether subscribing declares the descriptor’s expected topology first. Defaults to false: managing infrastructure is the user’s job, so creation is a deliberate opt-in.

When enabled, subscribing declares the bound exchanges (except the built-in amq.* ones and the default exchange), the queue, and the bindings.

Trait Implementations

impl Broker for LapinBroker

async fn connect(self) -> Result<Self::Connected, Self::Error>

Opens the connection and its shared publish channel, consuming the configuration.

Errors

Returns AmqpError::Connect when the URI cannot be parsed or the connection fails.

type Error = AmqpError

The error type returned by broker-level operations.

type Connected = ConnectedLapinBroker

The connected form of this broker: the typed witness that connect succeeded.

fn bindable(self) -> Bindable<Self>

where Self: 'static,

Wraps the broker for publisher-token minting before registration: the returned Bindable hands out Bound tokens via its bind, and is itself what with_broker takes.

impl Clone for LapinBroker

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 LapinBroker

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

Formats the value using the given formatter.

impl DescribeServer for LapinBroker

DescribeServer reports the configured AMQP address, which is what the AsyncAPI document records for the service.

The document is published and shared, so the credentials an AMQP URI carries must not reach it. ServerSpec::from_url drops them, along with the scheme and the vhost path.

fn describe_server(&self) -> ServerSpec

Returns the server coordinates for this broker.

impl InProcess for LapinBroker

Available on crate feature testing only.

The in-process mode: the connected form a test runs the production app against, carrying the in-process transport in place of the connection and every setting of this broker.

The URI is parsed as connect parses it, so a broker a service could not connect is not one a test can connect either.

fn connect_in_process( self, ) -> impl Future<Output = Result<Self::Connected, Self::Error>> + Send

Connects to the in-process transport, consuming the configuration.