Trait PayloadRow 

pub trait PayloadRow: QueueRow {
    type Column: 'static;

    // Required method
    fn payload(&self) -> &[u8] ;
}

A queue row that carries its message as bytes: payload mode.

The derive implements it for a struct with a #[field(payload)] field; a table described by hand implements it where its description sets InboxSpec::payload. A subscription hands the handler the payload decoded by a codec or a Deserialized type, as on any broker, and the bytes are lent from the row without a copy. Routes and by-name subscriptions read tables in this mode.

A struct without a payload field is in row mode: its handler takes the row itself, and the derive’s insert or a Repository over the service’s own Publish writes its tasks (see Row mode).

Examples

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

// `body` holds the message: `Job` is in payload mode, so a route can lead a name to it.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "jobs")]
pub struct Job {
    #[field(id, generated)]
    id: i64,
    #[field(group)]
    name: String,
    #[field(payload)]
    body: Vec<u8>,
}

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

#[derive(Deserialize)]
struct Email {
    to: String,
}

// The codec decodes `Email` from the bytes lent from `body`.
#[subscriber("emails")]
async fn send(email: &Email) -> HandlerOutcome {
    tracing::info!(to = %email.to, "sending");
    HandlerOutcome::ack()
}

pub fn app(pool: PgPool) -> RustStream {
    RustStream::new(AppInfo::new("mailer", "0.1.0")).with_broker(
        SqlxBroker::new(pool).route::<Job>("emails"),
        |b| {
            b.include(send);
        },
    )
}

A table described by hand names the payload column’s type, which a by-name subscription reads it in:

use chrono::{DateTime, Utc};
use ruststream_sqlx::dialect::Column;
use ruststream_sqlx::prelude::*;
use ruststream_sqlx::spec::{Lease, Payload};
use ruststream_sqlx::{InboxSpec, InboxTable, PayloadRow};
use serde::Deserialize;
use sqlx::{Sqlite, SqlitePool};

#[derive(sqlx::FromRow)]
pub struct Job {
    id: i64,
    body: Vec<u8>,
}

impl InboxTable for Job {
    type Id = i64;
    type Table = InboxSpec<(Lease<DateTime<Utc>>, Payload)>;
    const TABLE: Self::Table = InboxSpec::new("jobs", Column::new("id").generated())
        .lease(Column::new("locked_until"))
        .group(Column::new("name"))
        .payload(Column::new("body"));

    fn id(&self) -> &i64 {
        &self.id
    }
}

impl PayloadRow for Job {
    type Column = Vec<u8>;

    fn payload(&self) -> &[u8] {
        &self.body
    }
}

#[derive(Deserialize)]
struct Email {
    to: String,
}

// The codec decodes `Email` from the bytes lent from `body`.
#[subscriber(InboxQueue::<Job>::new("emails"))]
async fn send(email: &Email) -> HandlerOutcome {
    tracing::info!(to = %email.to, "sending");
    HandlerOutcome::ack()
}

pub fn app(pool: SqlitePool) -> RustStream {
    RustStream::new(AppInfo::new("mailer", "0.1.0")).with_broker(
        SqlxBroker::<Sqlite>::new(pool),
        |b| {
            b.include(send);
        },
    )
}

Required Associated Types

type Column: 'static

The payload column’s type, as the row decodes it: Vec<u8> or String, which a by-name subscription reads and writes the column as.

Required Methods

fn payload(&self) -> &[u8]

The message bytes, lent from the row.

Dyn Compatibility

This trait is not dyn compatible.

In older versions of Rust, dyn compatibility was called "object safety".

Implementors