Trait TimeSource

Source

pub trait TimeSource:
    Send
    + Sync
    + 'static {
    const DATABASE: bool;

    // Required method
    fn now() -> Option<SystemTime>;
}

Expand description

Where a table reads “now”: a Clock on the host, or DatabaseClock.

§Examples

use std::time::Duration;

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

// Each table names where it reads "now": `Charge` the host's clock, the default, and
// `MonthlyStatement` the database's.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "charge_jobs")]
pub struct Charge {
    #[field(id, generated)]
    id: i64,
    #[field(retry_after, generated)]
    retry_after: DateTime<Utc>,
    #[field(payload)]
    payload: Vec<u8>,
}

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

#[derive(Deserialize)]
struct Account {
    id: u64,
}

#[subscriber(InboxQueue::<Charge>::new("charges"))]
async fn charge(account: &Account) -> HandlerOutcome {
    tracing::info!(account.id, "charging");
    HandlerOutcome::retry_after(Duration::from_secs(60))
}

#[subscriber(InboxQueue::<MonthlyStatement>::new("statements"))]
async fn send_statement(account: &Account) -> HandlerOutcome {
    tracing::info!(account.id, "sending the statement");
    HandlerOutcome::retry_after(Duration::from_secs(60))
}

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