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; row mode shows one.

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
    }
}

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