Trait QueueTime

Source

pub trait QueueTime:
    Copy
    + Debug
    + Send
    + Sync
    + 'static {
    // Required methods
    fn from_system(at: SystemTime) -> Self;
    fn after(self, delay: Duration) -> Self;
    fn rounded_up(self) -> Self;
}

Expand description

A time a queue column holds: retry_after, processed_at or locked_until.

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. chrono::DateTime<Utc> implements it under the chrono feature and time::OffsetDateTime under time.

SQLite keeps times as text, and the 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 such a table may end up to a second late, and a delayed retry may come back up to a second early or late.

§Examples

use std::time::Duration;

use chrono::{DateTime, Utc};
use ruststream_sqlx::prelude::*;
use serde::Deserialize;
use sqlx::PgPool;

// `retry_after` holds a `DateTime<Utc>`: the crate writes a delayed retry in that type, as the
// service's own inserts write the column.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "webhook_jobs")]
pub struct Webhook {
    #[field(id, generated)]
    id: i64,
    #[field(retry_after, generated)]
    retry_after: DateTime<Utc>,
    #[field(payload)]
    payload: Vec<u8>,
}

#[derive(Deserialize)]
struct Call {
    url: String,
}

#[subscriber(InboxQueue::<Webhook>::new("webhooks"))]
async fn notify(call: &Call) -> HandlerOutcome {
    if post(&call.url).await {
        return HandlerOutcome::ack();
    }
    HandlerOutcome::retry_after(Duration::from_secs(30))
}

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

Required Methods§

Source

fn from_system(at: SystemTime) -> Self

The time at, in this type.

Source

fn after(self, delay: Duration) -> Self

This time moved delay later; the latest time the type holds when that overflows.

Source

fn rounded_up(self) -> Self

This time rounded up to the next whole second; a whole second stays, and so does the latest time the type holds.

A lease ends on a whole second: every temporal column stores one exactly, so the expiry a claim writes reads back unchanged as the delivery’s ownership token.