Trait Fetch 

pub trait Fetch<DB: Database>: InboxRow {
    // Required method
    fn fetch(
        conn: &mut DB::Connection,
        ids: &[Self::Id],
    ) -> impl Future<Output = Result<Vec<Self>, Error>> + Send;
}

Reads the rows of claimed ids, on the claim’s connection.

The dialect builds it, for a flat table and for a message assembled from header fields; a table names fetch as its own (custom(fetch), or own::Fetch by hand) to assemble messages itself, from other tables (a fetch over a join). Rows are matched to the claimed ids by their id field; a claimed id with no row settles by the decode-failure policy before its handler runs, and the log names the id. It runs inside the claim’s transaction, or right after a lease claim that leased the rows and committed in one statement, as on Postgres and SQLite. In the advisory lock form it runs once per row, with that row’s id alone, after the take that counted its attempt. Either way a delivery reports the attempt its row held before the claim. It runs only after a claim that took ids, so ids is never empty.

Examples

use ruststream_sqlx::{Fetch, Inbox};
use sqlx::{PgConnection, Postgres};

/// A job whose message is the body of the order it refers to.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "jobs", custom(fetch))]
pub struct Job {
    #[field(id)]
    id: i64,
    #[field(payload)]
    payload: Vec<u8>,
}

impl Fetch<Postgres> for Job {
    async fn fetch(conn: &mut PgConnection, ids: &[i64]) -> Result<Vec<Self>, sqlx::Error> {
        sqlx::query_as(
            "SELECT j.id, o.body AS payload FROM jobs j JOIN orders o ON o.id = j.order_id \
             WHERE j.id = ANY($1)",
        )
        .bind(ids)
        .fetch_all(conn)
        .await
    }
}

In the headers layout two messages may flatten one headers struct: the default fetch reads one from the queue table, and the service’s own fetch reads the other over a join. A claimed id the join finds no row for settles by the decode policy before the handler runs:

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

// order_jobs: job_id BIGSERIAL PRIMARY KEY, name TEXT, attempt SMALLINT DEFAULT 1, tenant TEXT,
// trace TEXT NULL, order_id BIGINT, note TEXT NULL
// orders: id BIGINT PRIMARY KEY, customer TEXT, total BIGINT
/// The queue table: the columns that run the queue, and the headers `tenant`, `trace` and
/// `order_id`.
#[derive(Debug, Clone, InboxHeaders, sqlx::FromRow)]
#[inbox(table = "order_jobs")]
pub struct OrderHeaders {
    #[field(id, generated)]
    job_id: i64,
    #[field(group)]
    name: String,
    #[field(attempt, generated)]
    attempt: i16,
    tenant: String,
    trace: Option<String>,
    order_id: i64,
}

/// A job and its note, read from the queue table by the default fetch.
#[derive(Debug, Clone, Inbox, sqlx::FromRow)]
pub struct OrderJob {
    #[field(headers)]
    #[sqlx(flatten)]
    headers: OrderHeaders,
    note: Option<String>,
}

/// 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))]
pub struct OrderMail {
    #[field(headers)]
    #[sqlx(flatten)]
    headers: OrderHeaders,
    customer: String,
    total: i64,
}

impl Fetch<Postgres> for OrderMail {
    async fn fetch(conn: &mut PgConnection, ids: &[i64]) -> Result<Vec<Self>, sqlx::Error> {
        sqlx::query_as(
            "SELECT j.job_id, j.name, j.attempt, j.tenant, j.trace, j.order_id, \
                    o.customer, o.total \
             FROM order_jobs j JOIN orders o ON o.id = j.order_id \
             WHERE j.job_id = ANY($1)",
        )
        .bind(ids)
        .fetch_all(conn)
        .await
    }
}

#[derive(Deserialize)]
pub struct Tenant {
    tenant: String,
    trace: Option<String>,
}

#[subscriber(InboxQueue::<OrderJob>::new("packing"))]
async fn pack(job: &OrderJob) -> HandlerOutcome {
    tracing::info!(attempt = job.headers.attempt, note = ?job.note, "packing");
    HandlerOutcome::ack()
}

// The header map is built from `OrderHeaders` on its first read, here by `Headers<Tenant>`.
#[subscriber(InboxQueue::<OrderMail>::new("receipts"))]
async fn send_receipt(mail: &OrderMail, Headers(tenant): Headers<Tenant>) -> HandlerOutcome {
    tracing::info!(
        tenant = %tenant.tenant,
        trace = ?tenant.trace,
        customer = %mail.customer,
        total = mail.total,
        "sending the receipt"
    );
    HandlerOutcome::ack()
}

pub fn app(pool: PgPool) -> RustStream {
    RustStream::new(AppInfo::new("shop", "0.1.0")).with_broker(SqlxBroker::new(pool), |b| {
        b.include(pack);
        // A job whose order is gone fails to decode: the default policy drops it.
        b.include(send_receipt);
    })
}

Required Methods

fn fetch( conn: &mut DB::Connection, ids: &[Self::Id], ) -> impl Future<Output = Result<Vec<Self>, Error>> + Send

The rows of ids, in any order.

Errors

The database’s error; the claim fails and its rows return to the queue.

Dyn Compatibility

This trait is not dyn compatible.

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

Implementors