Struct RoutedPublisher 

pub struct RoutedPublisher<DB: Database> { /* private fields */ }

The live form of Routed: publishes into the table the route table leads each name to.

Examples

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

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

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

#[derive(Serialize, Outgoing)]
#[outgoing(name = "orders")]
pub struct Order {
    id: u64,
}

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

    // The HTTP task owns this `RoutedPublisher`: each order becomes a row of `jobs`.
    let publisher = running.publisher(orders).await?;
    publisher.message(&Order { id: 7 }).publish().await?;

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

Trait Implementations

impl<DB: Database> Debug for RoutedPublisher<DB>

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

Formats the value using the given formatter.

impl<DB: QueueDatabase> Publisher for RoutedPublisher<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.