Trait Insert 

pub trait Insert<Connection>: Sync {
    // Required method
    fn insert<'c>(
        &'c self,
        conn: &'c mut Connection,
    ) -> impl Future<Output = Result<(), Error>> + Send + 'c;
}

Inserts the row: every field except the generated ones, the queue’s mechanics included, so a delayed task is a row whose retry_after lies ahead.

The derive implements it for the connection of each built-in dialect, PgConnection (feature postgres), MySqlConnection (feature mysql) and SqliteConnection (feature sqlite), with the statements built at compile time. A headers struct gets it from InboxHeaders: the queue table’s columns. A struct with a #[sqlx(flatten)] field gets none: the derive cannot see the nested struct’s columns, so the service writes that insert itself. A table described by hand implements it over the text dialect::insert renders from its description, binding the columns in the description’s order, as below.

Examples

use ruststream::OutgoingMessage;
use ruststream_sqlx::{Inbox, Insert, Publish};
use sqlx::{PgConnection, Postgres};

#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "email_jobs")]
pub struct SendEmail {
    #[field(id, generated)]
    job_id: i64,
    #[field(group)]
    name: String,
    #[field(payload)]
    payload: Vec<u8>,
}

impl Publish<Postgres> for SendEmail {
    async fn publish(
        conn: &mut PgConnection,
        message: &OutgoingMessage<'_>,
    ) -> Result<(), sqlx::Error> {
        let job = SendEmail {
            job_id: 0,
            name: message.name().to_owned(),
            payload: message.payload().to_vec(),
        };
        // `job_id` is generated: the database fills it in.
        job.insert(conn).await
    }
}

A table described by hand in row mode writes its insert over the text dialect::insert renders, in a constant:

use ruststream::runtime::{Input, SoloCarried};
use ruststream_sqlx::dialect::Column;
use ruststream_sqlx::dialect::insert::{self, Sql};
use ruststream_sqlx::prelude::*;
use ruststream_sqlx::spec::Attempt;
use ruststream_sqlx::{AttemptRow, InboxSpec, InboxTable};
use serde::{Deserialize, Serialize};
use sqlx::{PgConnection, PgPool};

#[derive(sqlx::FromRow, Clone)]
pub struct SendEmail {
    job_id: i64,
    name: String,
    attempt: i16,
    to: String,
    subject: Option<String>,
    #[sqlx(json)]
    attachments: Vec<Attachment>,
}

#[derive(Clone, Serialize, Deserialize)]
pub struct Attachment {
    file: String,
}

impl InboxTable for SendEmail {
    type Id = i64;
    type Table = InboxSpec<(Attempt,)>;
    const TABLE: Self::Table = InboxSpec::new("email_jobs", Column::new("job_id").generated())
        .group(Column::new("name"))
        .attempt(Column::new("attempt").generated())
        .data(&[Column::new("to"), Column::new("subject"), Column::new("attachments")]);

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

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

impl AttemptRow for SendEmail {
    type Attempt = i16;

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

// INSERT INTO "email_jobs" ("name", "to", "subject", "attachments") VALUES ($1, $2, $3, $4)
const INSERT: Sql<128> = insert::postgres(&SendEmail::TABLE.spec());

impl Insert<PgConnection> for SendEmail {
    async fn insert(&self, conn: &mut PgConnection) -> Result<(), sqlx::Error> {
        sqlx::query(INSERT.as_str())
            .bind(&self.name)
            .bind(&self.to)
            .bind(&self.subject)
            .bind(sqlx::types::Json(&self.attachments))
            .execute(conn)
            .await?;
        Ok(())
    }
}

Required Methods

fn insert<'c>( &'c self, conn: &'c mut Connection, ) -> impl Future<Output = Result<(), Error>> + Send + 'c

Inserts the row on conn.

Errors

The database’s error.

Examples

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

#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "signup_jobs")]
pub struct Signup {
    #[field(id, generated)]
    id: i64,
    #[field(payload)]
    payload: Vec<u8>,
}

/// A welcome mail to send, in row mode: the handler of `welcome_jobs` takes the row itself.
#[derive(Debug, Clone, Inbox, sqlx::FromRow)]
#[inbox(table = "welcome_jobs")]
pub struct Welcome {
    #[field(id, generated)]
    id: i64,
    email: String,
}

#[derive(Deserialize)]
struct Account {
    email: String,
}

// The welcome row commits with the signup's acknowledgement, or neither does.
#[subscriber(InboxQueue::<Signup>::new("signups"))]
async fn open_account(account: &Account, Ctx(mut tx): Ctx<keys::Tx<Postgres>>) -> HandlerOutcome {
    let welcome = Welcome { id: 0, email: account.email.clone() };
    match welcome.insert(&mut *tx).await {
        Ok(()) => HandlerOutcome::ack(),
        Err(_) => HandlerOutcome::retry(),
    }
}

pub fn app(pool: PgPool) -> RustStream {
    RustStream::new(AppInfo::new("accounts", "0.1.0")).with_broker(SqlxBroker::new(pool), |b| {
        b.include(open_account.transactional());
    })
}

Dyn Compatibility

This trait is not dyn compatible.

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

Implementors