Payload

A table with a payload column is in payload mode: the claim lends the column's bytes to the codec, and the handler takes the decoded message.

#[field(payload)] marks the column, by hand .payload(..) with PayloadRow, which names the column's type and lends its bytes. A claim reads the row, and the delivery lends those bytes to the codec as they are, with no copy. The codec is the mount's: JSON by default, or any codec of the core the router or the scope names. The bytes are the service's own: binary and empty payloads reach the handler byte for byte, and a publish through a route writes them as the codec encoded them.

A row whose bytes do not decode into the handler's type is settled by the subscription's decode policy before the handler runs. A handler that takes a row instead of a message mounts only on a table in row mode.

Example

A signup queue in payload mode: the handler takes the decoded Signup, and the acknowledged row keeps the bytes it was decoded from.

Macros
Manual

examples/sqlx_messages_payload.rs
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "event_jobs")]
struct EventJob {
    #[field(id, generated)]
    id: i64,
    #[field(processed_at)]
    processed_at: Option<DateTime<Utc>>,
    #[field(locked_until)]
    locked_until: Option<DateTime<Utc>>,
    // The message bytes, decoded by the codec of the mount: JSON unless it names another.
    #[field(payload)]
    payload: Vec<u8>,
}

#[subscriber(InboxQueue::<EventJob>::new("signups"))]
async fn welcome(signup: &Signup) -> HandlerOutcome {
    tracing::info!(email = %signup.email, "welcoming");
    HandlerOutcome::ack()
}

examples/manual/sqlx_messages_payload.rs
#[derive(sqlx::FromRow)]
struct EventJob {
    id: i64,
    payload: Vec<u8>,
}

impl InboxTable for EventJob {
    type Id = i64;
    type Table = InboxSpec<(ProcessedAt<DateTime<Utc>>, Lease<DateTime<Utc>>, Payload)>;
    const TABLE: Self::Table = InboxSpec::new("event_jobs", Column::new("id").generated())
        .processed_at(Column::new("processed_at"))
        .lease(Column::new("locked_until"))
        .payload(Column::new("payload"));

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

// The message bytes, lent to the codec of the mount: JSON unless it names another.
impl PayloadRow for EventJob {
    type Column = Vec<u8>;

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

struct Welcome;

impl Handle<Signup> for Welcome {
    fn handle(
        &self,
        signup: &Signup,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(email = %signup.email, "welcoming");
        ready(Ok(()))
    }
}

Example​

Example