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. It returns once the server has ended every claim the runtime cut short as it stopped the subscriptions. 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.

A claim the runtime cut short when it stopped a subscription may still run on the server: its connection waits for the server to end the claim’s statement and roll its transaction back, then closes, and shutdown waits for it, so the stopped broker holds no lock on the table.

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.

A pool the broker built (from_url) is closed last: it lends no new connection, and a delivery in work keeps the one it holds, which closes when it comes back. A settlement after shutdown that needs a new connection, as in the lease form, then fails, and its row comes back when its lease runs out. The service’s own pool (new) stays open.

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, after the evicted ones. 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 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 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.