Insert

Insert writes a task with the derive's generated insert, inside the service's own transaction, so the task commits with the data it serves.

The derive implements Insert for the connection of each enabled built-in driver: PgConnection, MySqlConnection and SqliteConnection. insert(&self, conn) writes every field of the struct but the generated ones: the roles, then the data columns, in the order the description lists them. A delayed task is a row whose retry_after lies ahead. The statement is built at compile time, and a call binds the fields and runs it on the connection it is given: a transaction of the service's own, or a handler's transaction. A headers struct gets the insert, and the message that flattens it gets none.

A table described by hand writes its insert over the text dialect::insert renders from its description, in a constant of type Sql<N>. The text names every column the database does not fill, in the description's order, and the impl binds them in that order. A capacity N the text outgrows is a build error that names the size to raise it to.

A row written this way wakes no subscription in the process: the subscription finds it on its next claim, at most a poll interval later. A publish through a route or a repository wakes the subscriptions of its table at once.

Example

An order and the task that sends its receipt go into one transaction: both commit, or neither does.

Macros
Manual

examples/sqlx_tables_insert.rs
use sqlx::SqlitePool;

#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "receipt_jobs")]
struct SendReceipt {
    #[field(id, generated)]
    id: i64,
    #[field(locked_until)]
    locked_until: Option<DateTime<Utc>>,
    #[field(payload)]
    payload: Vec<u8>,
}

/// Places an order and queues its receipt in one transaction: both commit, or neither does.
async fn place_order(pool: &SqlitePool, order: &Order) -> Result<(), Box<dyn Error>> {
    let mut tx = pool.begin().await?;
    sqlx::query("INSERT INTO orders (id, total) VALUES (?, ?)")
        .bind(order.id)
        .bind(order.total)
        .execute(&mut *tx)
        .await?;
    let receipt = SendReceipt {
        id: 0,
        locked_until: None,
        payload: serde_json::to_vec(&Receipt { order: order.id })?,
    };
    // `id` is generated: the insert leaves it to the database.
    receipt.insert(&mut *tx).await?;
    tx.commit().await?;
    Ok(())
}

examples/manual/sqlx_tables_insert.rs
use ruststream_sqlx::dialect::Column;
use ruststream_sqlx::dialect::insert::{self, Sql};
use ruststream_sqlx::spec::{Lease, Payload};
use ruststream_sqlx::{InboxSpec, InboxTable, PayloadRow};
use sqlx::{SqliteConnection, SqlitePool};

#[derive(sqlx::FromRow)]
struct SendReceipt {
    id: i64,
    payload: Vec<u8>,
}

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

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

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

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

// INSERT INTO `receipt_jobs` (`locked_until`, `payload`) VALUES (?, ?)
const INSERT: Sql<128> = insert::sqlite(&SendReceipt::TABLE.spec());

impl Insert<SqliteConnection> for SendReceipt {
    async fn insert(&self, conn: &mut SqliteConnection) -> Result<(), sqlx::Error> {
        sqlx::query(INSERT.as_str())
            .bind(None::<DateTime<Utc>>)
            .bind(&self.payload)
            .execute(conn)
            .await?;
        Ok(())
    }
}

/// Places an order and queues its receipt in one transaction: both commit, or neither does.
async fn place_order(pool: &SqlitePool, order: &Order) -> Result<(), Box<dyn Error>> {
    let mut tx = pool.begin().await?;
    sqlx::query("INSERT INTO orders (id, total) VALUES (?, ?)")
        .bind(order.id)
        .bind(order.total)
        .execute(&mut *tx)
        .await?;
    let receipt = SendReceipt {
        id: 0,
        payload: serde_json::to_vec(&Receipt { order: order.id })?,
    };
    receipt.insert(&mut *tx).await?;
    tx.commit().await?;
    Ok(())
}

Example​

Example