Trait Ack 

pub trait Ack<DB: Database>: OutboxRow {
    // Required method
    fn ack(
        conn: &mut DB::Connection,
        id: &Self::Id,
    ) -> impl Future<Output = Result<(), Error>> + Send;
}

Marks the record processed once its handler acknowledged the message.

The default sets processed_at from the database’s clock, or deletes the record when the struct has no processed_at field.

Examples

use ruststream_sqlx::{Outbox, outbox};

#[derive(Outbox, sqlx::FromRow)]
#[outbox(table = "outbox", custom(ack))]
pub struct OrderOutbox {
    #[field(id)]
    id: i64,
    #[field(name)]
    name: String,
    #[field(payload)]
    payload: Vec<u8>,
    #[field(processed_at)]
    processed_at: Option<DateTime<Utc>>,
}

// A processed record moves to the archive.
impl outbox::Ack<Postgres> for OrderOutbox {
    async fn ack(conn: &mut PgConnection, id: &i64) -> sqlx::Result<()> {
        sqlx::query(
            "WITH done AS (DELETE FROM outbox WHERE id = $1 RETURNING id, name, payload) \
             INSERT INTO outbox_archive (id, name, payload) SELECT id, name, payload FROM done",
        )
        .bind(id)
        .execute(conn)
        .await?;
        Ok(())
    }
}

pub fn app(pool: PgPool) -> impl App {
    let tracking = outbox! { pool: pool, "orders" => OrderOutbox };
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .layer(tracking.layer())
        .publish_layer(tracking.publish_layer())
        .with_broker(MemoryBroker::new(), |b| {
            b.include(place).out_reply(Publish);
            b.include(fulfil);
            b.after_startup(Publish, tracking.republish());
        })
}

Required Methods

fn ack( conn: &mut DB::Connection, id: &Self::Id, ) -> impl Future<Output = Result<(), Error>> + Send

Runs the ack event for the record id after its handler finished.

Errors

The database’s error; the record stays unprocessed, and the next startup publishes it again.

Dyn Compatibility

This trait is not dyn compatible.

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

Implementors