Rows

A table without a payload column is in row mode: a handler takes the struct itself as the driver read it, one row or a slice per claim.

A struct without a #[field(payload)] field, or a description without .payload(..), is in row mode. Its handler takes &Row, lent by the delivery with no codec and no copy. A batch handler takes &[Row], the rows of one claim in claim order, as a RowBatch. The struct derives Clone, because the test harness keeps a copy of each value, and does not derive Deserialize: a type that deserializes rides the codec. Either mistake is a compile error.

Every form serves row mode, and so does transactional mode, where the handler takes &Row beside Ctx<keys::Tx<DB>>. A handler of one row mounts in every form the core offers, a reply and Out slots included; a batch handler of rows mounts in the plain form. The column types are the service's to get right, and a row the driver cannot read settles by the decode policy. A nullable column takes Option<T>. The headers column becomes the delivery's headers, and the row the handler reads holds that column empty.

By hand, the description leaves .payload(..) out, names its own columns with .data(..), and the struct writes one line of the core's: impl Input for Row { type Axis = SoloCarried<Self>; }. A row-mode task is written with insert or a Publish of the service's own. In a test, with_value(&row) compares the row a handler took, and received_values::<Row>() returns every row the subscription lent.

Example

The mail queue has no payload column: the handler takes the Mail row, its recipient and subject read as columns.

Macros
Manual

examples/sqlx_messages_rows.rs
/// One mail to send: the table has no payload column, so the handler takes the row.
#[derive(Debug, Clone, Inbox, sqlx::FromRow)]
#[inbox(table = "mail_jobs")]
struct Mail {
    #[field(id, generated)]
    id: i64,
    #[field(attempt, generated)]
    attempt: i16,
    #[field(locked_until)]
    locked_until: Option<DateTime<Utc>>,
    recipient: String,
    subject: Option<String>,
}

#[subscriber(InboxQueue::<Mail>::new("mail"))]
async fn send(mail: &Mail) -> HandlerOutcome {
    tracing::info!(to = %mail.recipient, subject = ?mail.subject, mail.attempt, "sending");
    HandlerOutcome::ack()
}

examples/manual/sqlx_messages_rows.rs
use ruststream::runtime::SoloCarried;
use ruststream_sqlx::dialect::Column;
use ruststream_sqlx::spec::{Attempt, Lease};
use ruststream_sqlx::{AttemptRow, InboxSpec, InboxTable};

#[derive(Debug, Clone, sqlx::FromRow, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Mail {
    id: i64,
    attempt: i16,
    recipient: String,
    subject: Option<String>,
}

impl InboxTable for Mail {
    type Id = i64;
    type Table = InboxSpec<(Attempt, Lease<DateTime<Utc>>)>;
    const TABLE: Self::Table = InboxSpec::new("mail_jobs", Column::new("id").generated())
        .attempt(Column::new("attempt").generated())
        .lease(Column::new("locked_until"))
        .data(&[Column::new("recipient"), Column::new("subject")]);

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

// Row mode: a handler takes the row itself.
impl Input for Mail {
    type Axis = SoloCarried<Self>;
}

impl AttemptRow for Mail {
    type Attempt = i16;

    fn attempt(&self) -> &i16 {
        &self.attempt
    }
}

struct Sender;

impl Handle<Mail> for Sender {
    fn handle(
        &self,
        mail: &Mail,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(to = %mail.recipient, subject = ?mail.subject, mail.attempt, "sending");
        ready(Ok(()))
    }
}

Example​

Example