Fetch

custom(fetch) reads the claimed rows with the service's own SELECT, such as a join; a claimed id it returns no row for settles by the decode policy.

A message whose data lives in other tables lists fetch in #[inbox(custom(..))] and implements Fetch for its database; by hand the chain adds .own::<own::Fetch>(), with own::Fetch in its type. The fetch takes the claimed ids and returns the rows it found. Its SELECT returns the headers struct's columns under their names, locked_until included in the lease form, beside the message's own. The crate matches the rows to the claimed ids by the headers struct's id field, and an empty claim does not call the fetch.

The fetch runs on the claim's connection: in the claim's transaction in the row lock form, and once per row, with that row's id alone, in the advisory lock form. A subscription to such a message does not check the message's own columns at startup, because the fetch reads them from wherever they live.

A claimed id that the fetch returned no row for, such as a job whose order the join does not find, settles by the subscription's decode policy before the handler runs. The default policy drops the row, and the log names the id. A batch lends its handler the rows the fetch found, and the policy settles the others.

Example

Each claimed job is read joined to its order; SQLite takes the claimed ids as a JSON array, read back by json_each. A job whose order is gone joins no row, and the default policy drops it.

Macros
Manual

examples/sqlx_headers_fetch.rs
use ruststream_sqlx::Fetch;
use sqlx::types::Json;
use sqlx::{Sqlite, SqliteConnection};

/// A job and the order it is for, read by the service's own fetch over a join.
#[derive(Debug, Clone, Inbox, sqlx::FromRow)]
#[inbox(custom(fetch))]
struct OrderMail {
    #[field(headers)]
    #[sqlx(flatten)]
    headers: MailHeaders,
    customer: String,
    total: i64,
}

impl Fetch<Sqlite> for OrderMail {
    async fn fetch(conn: &mut SqliteConnection, ids: &[i64]) -> Result<Vec<Self>, sqlx::Error> {
        // SQLite takes a list of ids as a JSON array, read back by `json_each`.
        sqlx::query_as(
            "SELECT j.job_id, j.locked_until, j.order_id, o.customer, o.total \
             FROM mail_jobs j JOIN orders o ON o.id = j.order_id \
             WHERE j.job_id IN (SELECT value FROM json_each(?))",
        )
        .bind(Json(ids))
        .fetch_all(conn)
        .await
    }
}

examples/manual/sqlx_headers_fetch.rs
use ruststream_sqlx::spec::{self, Lease, own};
use ruststream_sqlx::{Fetch, InboxSpec, InboxTable};
use sqlx::types::Json;
use sqlx::{Sqlite, SqliteConnection};

#[derive(Debug, Clone, sqlx::FromRow, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct OrderMail {
    #[sqlx(flatten)]
    headers: MailHeaders,
    customer: String,
    total: i64,
}

impl InboxTable for OrderMail {
    type Id = i64;
    type Table = InboxSpec<(Lease<DateTime<Utc>>, spec::HeaderFields, own::Fetch)>;
    const TABLE: Self::Table = InboxSpec::new("mail_jobs", Column::new("job_id").generated())
        .lease(Column::new("locked_until"))
        .data(&[Column::new("order_id")])
        .fetching(&[Column::new("customer"), Column::new("total")])
        .header_fields()
        .own::<own::Fetch>();

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

impl Fetch<Sqlite> for OrderMail {
    async fn fetch(conn: &mut SqliteConnection, ids: &[i64]) -> Result<Vec<Self>, sqlx::Error> {
        // SQLite takes a list of ids as a JSON array, read back by `json_each`.
        sqlx::query_as(
            "SELECT j.job_id, j.locked_until, j.order_id, o.customer, o.total \
             FROM mail_jobs j JOIN orders o ON o.id = j.order_id \
             WHERE j.job_id IN (SELECT value FROM json_each(?))",
        )
        .bind(Json(ids))
        .fetch_all(conn)
        .await
    }
}

Example​

Example