Struct RepositoryPublisher 

pub struct RepositoryPublisher<DB: Database, Row> { /* private fields */ }

The live form of a Repository: publishes into Row’s table through its Publish.

After the broker shut down it answers SqlxBrokerError::Closed, never a success.

Examples

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

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

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

#[derive(Serialize, Outgoing)]
#[outgoing(name = "reports")]
pub struct Report {
    month: u32,
}

pub async fn run(pool: PgPool) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    let broker = SqlxBroker::new(pool).bindable();
    let reports = broker.bind(Repository::<ReportJob>::default());
    let running = RustStream::new(AppInfo::new("reports", "0.1.0"))
        .with_broker(broker, |_b| {})
        .start()
        .await?;

    // A task scheduled from outside any handler: the service's HTTP endpoint owns this
    // `RepositoryPublisher` and writes each request into `report_jobs`.
    let publisher = running.publisher(reports).await?;
    publisher.message(&Report { month: 10 }).publish().await?;

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

Trait Implementations

impl<DB: Database, Row> Debug for RepositoryPublisher<DB, Row>

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

Formats the value using the given formatter.

impl<DB, Row> Publisher for RepositoryPublisher<DB, Row>

where DB: QueueDatabase, Row: Publish<DB> + Events<DB>,

type Payload = Lend

How this transport consumes the payload: Lend where it reads the bytes and keeps nothing, Take where its client keeps them.

type Error = SqlxBrokerError

The error type returned by publish.

type Options = ()

The broker’s per-message settings: a QoS, a priority, an ordering key, whatever this transport lets one message differ from the next in.

async fn publish( &self, msg: OutgoingMessage<'_>, _options: Option<&()>, ) -> Result<(), SqlxBrokerError>

Publishes a message to the broker, with the per-message settings the call site adjusted.

fn base_headers(&self) -> Option<&HeaderMap>

The headers this publisher contributes to every message it sends, underneath whatever the message itself names.