Trait Clock

Source

pub trait Clock:
    Send
    + Sync
    + 'static {
    // Required method
    fn now() -> SystemTime;
}

Expand description

A clock on the host: what “now” is for the statements that bind it.

#[inbox(clock = MyClock)] makes a table read it. Hosts’ clocks must agree to well within the delays the queue works with; keeping them in step is the service’s concern (NTP).

§Examples

use std::time::{Duration, SystemTime, UNIX_EPOCH};

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

/// The host's time in whole seconds, as the service's `DATETIME(0)` columns keep it.
pub struct WholeSeconds;

impl Clock for WholeSeconds {
    fn now() -> SystemTime {
        let since = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default();
        UNIX_EPOCH + Duration::from_secs(since.as_secs())
    }
}

#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "reminder_jobs", clock = WholeSeconds)]
pub struct Reminder {
    #[field(id, generated)]
    id: i64,
    #[field(retry_after, generated)]
    retry_after: DateTime<Utc>,
    #[field(payload)]
    payload: Vec<u8>,
}

#[derive(Deserialize)]
struct Note {
    text: String,
}

#[subscriber(InboxQueue::<Reminder>::new("reminders"))]
async fn remind(note: &Note) -> HandlerOutcome {
    tracing::info!(text = %note.text, "reminding");
    // Back in an hour, counted from `WholeSeconds`.
    HandlerOutcome::retry_after(Duration::from_secs(3600))
}

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