Struct ConnectedLapinBroker 

pub struct ConnectedLapinBroker { /* private fields */ }

The typed witness that Broker::connect succeeded: holds the live connection.

Everything connection-bound hangs off this value: subscriptions (Subscribe, subscribe), publishers (publisher), and requesters (requester). ConnectedBroker::shutdown consumes it, so a publish or subscribe after shutdown is a compile error for the owner of the handle.

Implementations

impl ConnectedLapinBroker

pub fn server_spec(&self) -> ServerSpec

The AsyncAPI server description of the connection this broker dialled.

pub async fn subscribe( &self, def: impl QueueDescriptor, ) -> Result<LapinSubscriber, AmqpError>

Opens a subscription for def, declaring its topology first when the broker opted in.

Takes either queue descriptor: RabbitQueue or RabbitQuorumQueue.

Errors

Returns AmqpError::Closed after shutdown, AmqpError::Declare when opted-in declaration fails, AmqpError::InvalidOptions for contradictory descriptor options and for a retry declaration this queue cannot carry, and AmqpError::Subscribe when the channel or consumer cannot be opened (for example the queue, or the waiting queue or delayed exchange its delay names, does not exist and declaration was not opted into).

pub fn publisher<P: LapinPublishPolicy>(&self, policy: P) -> P::Live

A live publisher for policy.

LapinPublish pairs into the fire-and-forget publisher, ConfirmsPublish into the confirm-transactional one, and ServerTxPublish into the AMQP-server-transactional one. All three are cheap to build and cheap to clone.

Examples

use std::error::Error;

use ruststream::Broker;
use ruststream_lapin::prelude::*;
use serde::Serialize;

#[derive(Serialize, Outgoing)]
#[outgoing(name = "order.placed")]
struct OrderPlaced {
    id: u64,
}

// A one-off backfill outside any service: connect, publish, confirm.
async fn backfill(ids: &[u64]) -> Result<(), Box<dyn Error>> {
    let connected = LapinBroker::new("amqp://localhost:5672").connect().await?;
    let events = connected.publisher(LapinPublish::default().exchange("events").confirms());
    for &id in ids {
        events.message(&OrderPlaced { id }).publish().await?;
    }
    Ok(())
}

pub fn requester(&self, policy: LapinRequest) -> LapinRequester

A live request/reply client over RabbitMQ direct reply-to.

The requester half of LapinRequest; publisher accepts the same policy, this accessor only names the result.

Trait Implementations

impl ConnectedBroker for ConnectedLapinBroker

async fn shutdown(self) -> Result<Self::Closed, Self::Error>

Closes the connection, consuming the connected form.

Errors

Returns AmqpError::Connect when the close handshake fails.

type Error = AmqpError

The error type returned by connected-broker operations.

type Closed = ClosedLapinBroker

The terminal witness returned by shutdown.

impl Debug for ConnectedLapinBroker

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

Formats the value using the given formatter.

impl DefaultPublish for ConnectedLapinBroker

type Policy = LapinPublish

The broker’s plain publish policy, constructible with its defaults.

impl PublishPolicy<ConnectedLapinBroker> for LapinPublish

type Live = LapinPublisher

The live form this policy pairs into: a Publisher for a leaf policy, or the live wiring form for a combinator stack (a typed publisher over a policy pairs into the same typed publisher over the live leaf).

fn pair( self, connected: &ConnectedLapinBroker, ) -> impl Future<Output = Result<Self::Live, PairError>>

Pairs the policy with a connected broker, producing the live publisher.

fn channel_bindings(&self, destination: &str) -> Bindings

What a publish through this policy adds to its channel in the generated AsyncAPI document.

fn operation_bindings(&self, _destination: &str) -> Bindings

What a publish through this policy adds to its send operation in the document.

fn reply_address_location(&self) -> Option<&'static str>

Where a client finds the address of a reply this policy publishes, as a runtime expression.

fn message_bindings(&self, channel: &str) -> Bindings

What a publish through this policy adds to the messages that leave through it.

impl PublishPolicy<ConnectedLapinBroker> for ConfirmsPublish

type Live = ConfirmsPublisher

The live form this policy pairs into: a Publisher for a leaf policy, or the live wiring form for a combinator stack (a typed publisher over a policy pairs into the same typed publisher over the live leaf).

fn pair( self, connected: &ConnectedLapinBroker, ) -> impl Future<Output = Result<Self::Live, PairError>>

Pairs the policy with a connected broker, producing the live publisher.

fn channel_bindings(&self, destination: &str) -> Bindings

What a publish through this policy adds to its channel in the generated AsyncAPI document.

fn operation_bindings(&self, _destination: &str) -> Bindings

What a publish through this policy adds to its send operation in the document.

fn reply_address_location(&self) -> Option<&'static str>

Where a client finds the address of a reply this policy publishes, as a runtime expression.

fn message_bindings(&self, channel: &str) -> Bindings

What a publish through this policy adds to the messages that leave through it.

impl PublishPolicy<ConnectedLapinBroker> for ServerTxPublish

type Live = ServerTxPublisher

The live form this policy pairs into: a Publisher for a leaf policy, or the live wiring form for a combinator stack (a typed publisher over a policy pairs into the same typed publisher over the live leaf).

fn pair( self, connected: &ConnectedLapinBroker, ) -> impl Future<Output = Result<Self::Live, PairError>>

Pairs the policy with a connected broker, producing the live publisher.

fn channel_bindings(&self, destination: &str) -> Bindings

What a publish through this policy adds to its channel in the generated AsyncAPI document.

fn operation_bindings(&self, _destination: &str) -> Bindings

What a publish through this policy adds to its send operation in the document.

fn reply_address_location(&self) -> Option<&'static str>

Where a client finds the address of a reply this policy publishes, as a runtime expression.

fn message_bindings(&self, channel: &str) -> Bindings

What a publish through this policy adds to the messages that leave through it.

impl PublishPolicy<ConnectedLapinBroker> for LapinRequest

type Live = LapinRequester

The live form this policy pairs into: a Publisher for a leaf policy, or the live wiring form for a combinator stack (a typed publisher over a policy pairs into the same typed publisher over the live leaf).

fn pair( self, connected: &ConnectedLapinBroker, ) -> impl Future<Output = Result<Self::Live, PairError>>

Pairs the policy with a connected broker, producing the live publisher.

fn channel_bindings(&self, destination: &str) -> Bindings

What a publish through this policy adds to its channel in the generated AsyncAPI document.

fn operation_bindings(&self, _destination: &str) -> Bindings

What a publish through this policy adds to its send operation in the document.

fn reply_address_location(&self) -> Option<&'static str>

Where a client finds the address of a reply this policy publishes, as a runtime expression.

fn message_bindings(&self, channel: &str) -> Bindings

What a publish through this policy adds to the messages that leave through it.

impl RedeliveryAddressed<ConnectedLapinBroker> for RabbitQueue

The queue name: on the default exchange a routing key addresses the queue that carries it, so that is where the runtime’s retry copy reaches this subscription again.

The answer holds for a retry publisher on the default exchange, which is what LapinPublish is unless configured otherwise. A publisher aimed at a topic or direct exchange reaches the queue only through a binding under this name, so bind it there or leave the retry publisher on the default exchange.

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

Where a publish reaches this subscription again.

impl Subscribe for ConnectedLapinBroker

type Copies = AddressedCopies

A queue name is an address as well as a subscription: on the default exchange a routing key addresses the queue that carries it, so the service publishes a delayed redelivery back under the name it subscribed to.

The copy reaches the queue as long as the retry publisher sends on the default exchange, which is what LapinPublish does unless exchange says otherwise; pointing that publisher at a topic exchange with no binding under the queue name would send the copy nowhere, so bind it there or leave the retry publisher on the default exchange.

async fn subscribe(&self, name: &str) -> Result<Self::Subscriber, Self::Error>

Subscribes to the queue name with descriptor defaults (durable, shared).

type Subscriber = LapinSubscriber

The subscriber type opened by a by-name subscription.

fn declare_retry( &self, name: &str, declaration: &RetryDeclaration, ) -> Result<(), DeclareRetryError>

Takes what a registration mounted by a bare name declared about its retries, before the subscription opens.

impl SubscriptionSource<ConnectedLapinBroker> for RabbitQueue

type Copies = AddressedCopies

The service publishes the copies of a retry, and the queue name is where they go: on the default exchange a routing key addresses the queue that carries it.

A classic queue keeps no counter and has no delivery limit, so there is nothing on the server to end a retry loop; the runtime counts and the mount site names the publisher the copies leave through with out_retry(policy).

fn channel_bindings(&self) -> Bindings

The queue this subscription consumes, as the amqp binding names its settings.

fn operation_bindings(&self) -> Bindings

The consumer’s own half: this crate acknowledges by hand on every subscription.

type Subscriber = LapinSubscriber

The subscriber type this source opens.

fn name(&self) -> &str

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

async fn subscribe( self, connected: &ConnectedLapinBroker, ) -> Result<Self::Subscriber, AmqpError>

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

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 message_bindings(&self) -> Bindings

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

impl SubscriptionSource<ConnectedLapinBroker> for RabbitQuorumQueue

type Copies = BrokerMoves

The queue moves a spent delivery itself: its x-delivery-limit counts the deliveries and its dead-letter route takes the message that runs out of them, with nothing published by this process. That is what makes out_retry(policy) a compile error here, and what keeps the cap holding while the service is down.

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

Takes the registration’s cap and dead-letter destination into the descriptor, so the queue this broker declares carries them as topology.

fn channel_bindings(&self) -> Bindings

The queue this subscription consumes, as the amqp binding names its settings.

fn operation_bindings(&self) -> Bindings

The consumer’s own half: this crate acknowledges by hand on every subscription.

type Subscriber = LapinSubscriber

The subscriber type this source opens.

fn name(&self) -> &str

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

async fn subscribe( self, connected: &ConnectedLapinBroker, ) -> Result<Self::Subscriber, AmqpError>

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

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 message_bindings(&self) -> Bindings

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

impl TestableBroker for ConnectedLapinBroker

Available on crate feature testing only.

The harness’s view of the in-process transport: what it injects, what it reads back, and the coordinator it counts the in-flight deliveries with.

Panics

inject and published panic on a broker connected with connect: the harness drives only the connection connect_in_process produced, and a live connection has no log to read and no synchronous way to take a message.

fn routes(&self, destination: &str, subscriptions: &[&str]) -> Vec<usize>

A queue hands each message to one of its consumers, so a publish to a queue with two subscriptions is owed by one of them, never by both.

The broker’s default publisher sends on the default exchange, which routes by queue name, so the candidates are the subscriptions of destination’s name, in the order they opened. In process the answer is the consumer the transport’s rotation serves next. Against a server the pick is the server’s and not observable from here, so a live test opens one subscription per queue.

Against a server a publish through a named exchange is owed by the subscriptions whose queues the service bound to it: by the binding key on a direct exchange, by the pattern on a topic exchange, every one on a fanout exchange. A consistent-hash exchange hands a key to the one queue its hash picks, so every bound queue owes the publish until the first delivery of that key shows which one took it.

Panics

Panics against a server when destination has more than one subscription: the harness would wait on a consumer the server may not have picked, and time out without saying why.

fn install_coordinator(&self, coordinator: Coordinator)

Installs the harness coordinator into this broker’s bus for a test run. Idempotent: a second install on the same broker is ignored.

fn inject(&self, message: OutgoingMessage<'_>)

Injects a message onto the bus as an external producer would, synchronously (no awaiting). Routes through the broker’s normal fanout, so it is recorded and counted like any publish.

fn published(&self, name: &str) -> Vec<RawMessage>

Returns every message published to name on this broker, in publish order, after the evicted ones. Backs the harness’s published::<T>(name) assertions and expect_published.

fn evicted(&self, name: &str) -> usize

How many of the oldest messages published to name this broker no longer holds, so published starts after them.

fn backlog(&self) -> Backlog

What a subscription opened by name receives of the messages published to that name before it opened, as the real broker answers it: Backlog::Missed for publish/subscribe, Backlog::Delivered for a queue or a log that keeps them.