QueueTime

A time column holds chrono::DateTime<Utc> or time::OffsetDateTime; SQLite compares times as text, exact for chrono and to the second for time.

QueueTime is the type a retry_after, processed_at or locked_until column holds: chrono::DateTime<Utc> under the chrono feature, time::OffsetDateTime under time. A field of such a role is the time or an Option of it (TimeColumn), and an optional one holds NULL until the crate writes it. The crate reads "now" and moves it forward in the column's own type, so sqlx encodes it exactly as the service writes times itself. SQLite keeps times as text, and a claim compares them as text. The text sqlx writes for a chrono time, RFC 3339 with +00:00, sorts as the times themselves. The text it writes for a time value ends in Z and drops the fraction's trailing zeros, so it sorts two times right only when they fall in different seconds. On SQLite a lease of a time table may therefore end up to a second late, and a delayed retry may come back up to a second early or late. By hand, the time type is a parameter of the description's markers, as in Lease<OffsetDateTime> and ProcessedAt<OffsetDateTime>.

Example

A backup queue keeps its lease and its mark in the time crate's type, and the mark reaches the table as sqlx writes a time value.

Macros
Manual

examples/sqlx_time_types.rs
use ruststream_sqlx::prelude::*;
use sqlx::Connection;
use sqlx::sqlite::SqlitePoolOptions;
use time::OffsetDateTime;

/// One row of `backup_jobs`, with its times in the `time` crate's type.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "backup_jobs")]
struct Backup {
    #[field(id, generated)]
    id: i64,
    #[field(locked_until)]
    locked_until: Option<OffsetDateTime>,
    #[field(processed_at)]
    processed_at: Option<OffsetDateTime>,
    #[field(payload)]
    payload: Vec<u8>,
}

#[subscriber(InboxQueue::<Backup>::new("backups"))]
async fn back_up(volume: &Volume) -> HandlerOutcome {
    tracing::info!(name = %volume.name, "backing up");
    HandlerOutcome::ack()
}

/// The service on the database `url` names.
fn service(url: &str) -> RustStream {
    let schema_url = url.to_owned();
    // Nothing connects here: the broker builds its pool when the service starts.
    let broker = SqlxBroker::from_url(url, SqlitePoolOptions::new());
    RustStream::new(AppInfo::new("backups", "0.1.0"))
        // The table, created before the subscription checks it.
        .on_startup(async move |()| {
            let mut conn = SqliteConnection::connect(&schema_url).await?;
            sqlx::raw_sql(SCHEMA).execute(&mut conn).await?;
            conn.close().await
        })
        .with_broker(broker.route::<Backup>("backups"), |b| {
            b.include(back_up);
        })
}

examples/manual/sqlx_time_types.rs
use ruststream_sqlx::prelude::*;
use sqlx::Connection;
use sqlx::sqlite::SqlitePoolOptions;
use time::OffsetDateTime;

/// One row of `backup_jobs`, with its times in the `time` crate's type.
#[derive(sqlx::FromRow)]
struct Backup {
    id: i64,
    payload: Vec<u8>,
}

impl InboxTable for Backup {
    type Id = i64;
    type Table = InboxSpec<(Lease<OffsetDateTime>, ProcessedAt<OffsetDateTime>, Payload)>;
    const TABLE: Self::Table = InboxSpec::new("backup_jobs", Column::new("id").generated())
        .lease(Column::new("locked_until"))
        .processed_at(Column::new("processed_at"))
        .payload(Column::new("payload"));

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

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

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

struct BackUp;

impl Handle<Volume> for BackUp {
    fn handle(
        &self,
        volume: &Volume,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(name = %volume.name, "backing up");
        ready(Ok(()))
    }
}

/// The service on the database `url` names.
fn service(url: &str) -> RustStream {
    let schema_url = url.to_owned();
    // Nothing connects here: the broker builds its pool when the service starts.
    let broker = SqlxBroker::from_url(url, SqlitePoolOptions::new());
    RustStream::new(AppInfo::new("backups", "0.1.0"))
        // The table, created before the subscription checks it.
        .on_startup(async move |()| {
            let mut conn = SqliteConnection::connect(&schema_url).await?;
            sqlx::raw_sql(SCHEMA).execute(&mut conn).await?;
            conn.close().await
        })
        .with_broker(broker.route::<Backup>("backups"), |b| {
            let back_up =
                subscriber("backups", BackUp).map_source(|_| InboxQueue::<Backup>::new("backups"));
            b.include(back_up.build());
        })
}

Example​

Example