Struct ConnectedSqlxBroker 

pub struct ConnectedSqlxBroker<DB: Database, D = BuiltIn<DB>> { /* private fields */ }

The connected inbox broker: the typed witness that the database answered.

It hands out publishers and opens subscriptions; shutdown stops its claim loops and the extension of its leases, and refuses every handle it handed out. A delivery in work keeps its lease after shutdown and settles as before; its lease is no longer extended. In the advisory lock form shutdown releases the lock of every delivery in work and returns once no lock of the broker is left; such a delivery’s settlement then fails with SqlxBrokerError::Closed.

Examples

use ruststream::OutgoingMessage;
use ruststream_sqlx::prelude::*;
use serde::Serialize;
use sqlx::{PgConnection, PgPool, Postgres};

#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "cleanup_jobs")]
pub struct Cleanup {
    #[field(id, generated)]
    id: i64,
    #[field(payload)]
    payload: Vec<u8>,
}

impl Publish<Postgres> for Cleanup {
    async fn publish(
        conn: &mut PgConnection,
        message: &OutgoingMessage<'_>,
    ) -> Result<(), sqlx::Error> {
        sqlx::query("INSERT INTO cleanup_jobs (payload) VALUES ($1)")
            .bind(message.payload())
            .execute(conn)
            .await?;
        Ok(())
    }
}

#[derive(Serialize, Outgoing)]
#[outgoing(name = "cleanup")]
struct Sweep {
    older_than_days: u32,
}

pub fn app(pool: PgPool) -> RustStream {
    RustStream::new(AppInfo::new("maintenance", "0.1.0")).with_broker(SqlxBroker::new(pool), |b| {
        // Once the broker connected, the policy pairs against the `ConnectedSqlxBroker` and the
        // hook schedules the first sweep.
        b.after_startup(Repository::<Cleanup>::default(), async move |cleanups| {
            cleanups.message(&Sweep { older_than_days: 30 }).publish().await
        });
    })
}

Trait Implementations

impl<DB: QueueDatabase, D: Dialect + 'static> ConnectedBroker for ConnectedSqlxBroker<DB, D>

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

Stops the claim loops and the extension of every lease, and refuses every handle handed out before. A delivery in work in the lease or row lock form settles as before; its lease runs out unless it settles first.

In the advisory lock form it then releases the lock of every delivery in work: each session unlocks its key and goes back to the pool, or closes where the database does not confirm the release. It waits for each session lent to a settlement in flight, and for the sessions still closing after a delivery or a claim was dropped, and returns once no lock of the broker is left. In a service the runtime has stopped its subscriptions by then, and its shutdown timeout bounds the handlers that hold a session.

type Error = SqlxBrokerError

The error type returned by connected-broker operations.

type Closed = ClosedSqlxBroker

The terminal witness returned by shutdown.

impl<DB: Database, D: Dialect> Debug for ConnectedSqlxBroker<DB, D>

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

Formats the value using the given formatter.

impl<DB: QueueDatabase, D: Dialect + 'static> DefaultPublish for ConnectedSqlxBroker<DB, D>

type Policy = Routed

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

impl<DB, D, Row> PublishPolicy<ConnectedSqlxBroker<DB, D>> for Repository<Row>

where DB: QueueDatabase, D: Dialect + 'static, Row: Publish<DB> + Events<DB>,

type Live = RepositoryPublisher<DB, Row>

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: &ConnectedSqlxBroker<DB, D>, ) -> impl Future<Output = Result<Self::Live, PairError>> + Send

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

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

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

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

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

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

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

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

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

impl<DB: QueueDatabase, D: Dialect + 'static> PublishPolicy<ConnectedSqlxBroker<DB, D>> for Routed

type Live = RoutedPublisher<DB>

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: &ConnectedSqlxBroker<DB, D>, ) -> impl Future<Output = Result<Self::Live, PairError>> + Send

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

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

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

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

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

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

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

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

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

impl<DB, D> Subscribe for ConnectedSqlxBroker<DB, D>

where DB: QueueDatabase + RoleColumns, D: ByName<DB> + 'static,

type Subscriber = NamedSubscriber<DB, D>

The subscriber type opened by a by-name subscription.

type Copies = BrokerMoves

Who publishes the copies a retry of a by-name subscription is made of, and who names where they go. This is what the Name source reports, so it decides how #[subscriber("orders")] retries on this broker.

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

Opens a subscription to name, producing this broker’s Subscriber.

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<DB, D, Row, Mode> SubscriptionSource<ConnectedSqlxBroker<DB, D>> for InboxQueue<Row, Mode>

where DB: QueueDatabase, D: Dialect + Opens<Row::Opening> + 'static, Row: InboxRow + Events<DB>, Row::Form: FormOn<D>, Mode: InboxMode,

type Subscriber = InboxSubscriber<DB, Row, Mode>

The subscriber type this source opens.

type Copies = BrokerMoves

Who publishes the copies this subscription’s retries are made of, and who names where they go: AddressedCopies, NamedCopies or BrokerMoves.

fn name(&self) -> &str

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

async fn subscribe( self, connected: &ConnectedSqlxBroker<DB, D>, ) -> Result<Self::Subscriber, SqlxBrokerError>

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

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

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

fn channel_bindings(&self) -> Bindings

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

fn declare_retry_on( &self, _connected: &ConnectedSqlxBroker<DB, D>, 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.

impl<DB: QueueDatabase, D: Dialect + 'static> TestableBroker for ConnectedSqlxBroker<DB, D>

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. Backs the harness’s published::<T>(name) assertions and expect_published.

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.

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

Which subscriptions of this broker a message published to destination is delivered to, as the broker’s own routing picks them: the positions in subscriptions (the names the app subscribed under on this broker) of the ones it delivers to.