Clock

A clock of the service's own implements Clock and takes the place of SystemClock in clock = .. or .clock::<..>(), for every statement that reads "now".

Clock has one function, now() -> SystemTime, and the crate turns its answer into each column's own type. #[inbox(clock = WholeSeconds)] names the clock on the derive; by hand .clock::<WholeSeconds>() names it, and Clock<WholeSeconds> goes into the description's type. The clock is a host clock, so every form takes it, the lease form included. Such a clock fits a service whose other tables keep times in a form of their own, whole seconds here, or a host that has to read a time other than its system clock's. A clock that is a type parameter of the struct is not taken by checked. In process the table reads its own clock, as in production, and a paused tokio clock does not move it.

Example

The reminders' table reads the host's time in whole seconds, and an acknowledgement marks the row with it.

Macros
Manual

examples/sqlx_time_clock.rs
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use chrono::{DateTime, Utc};
use ruststream_sqlx::prelude::*;
use ruststream_sqlx::{Clock, SystemClock};
use sqlx::Connection;
use sqlx::sqlite::SqlitePoolOptions;

/// The host's time in whole seconds.
struct WholeSeconds;

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

/// One row of `reminder_jobs`, on the service's clock.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "reminder_jobs", clock = WholeSeconds)]
struct Remind {
    #[field(id, generated)]
    id: i64,
    #[field(locked_until)]
    locked_until: Option<DateTime<Utc>>,
    #[field(processed_at)]
    processed_at: Option<DateTime<Utc>>,
    #[field(payload)]
    payload: Vec<u8>,
}

#[subscriber(InboxQueue::<Remind>::new("reminders"))]
async fn remind(note: &Note) -> HandlerOutcome {
    tracing::info!(text = %note.text, "reminding");
    // `processed_at` takes `WholeSeconds::now()`.
    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("reminders", "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::<Remind>("reminders"), |b| {
            b.include(remind);
        })
}

examples/manual/sqlx_time_clock.rs
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use chrono::{DateTime, Utc};
use ruststream_sqlx::prelude::*;
use ruststream_sqlx::{Clock, SystemClock, spec};
use sqlx::Connection;
use sqlx::sqlite::SqlitePoolOptions;

/// The host's time in whole seconds.
struct WholeSeconds;

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

/// One row of `reminder_jobs`, on the service's clock.
#[derive(sqlx::FromRow)]
struct Remind {
    id: i64,
    payload: Vec<u8>,
}

impl InboxTable for Remind {
    type Id = i64;
    type Table = InboxSpec<(
        spec::Clock<WholeSeconds>,
        Lease<DateTime<Utc>>,
        ProcessedAt<DateTime<Utc>>,
        Payload,
    )>;
    const TABLE: Self::Table = InboxSpec::new("reminder_jobs", Column::new("id").generated())
        .clock::<WholeSeconds>()
        .lease(Column::new("locked_until"))
        .processed_at(Column::new("processed_at"))
        .payload(Column::new("payload"));

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

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

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

struct RemindHandler;

impl Handle<Note> for RemindHandler {
    fn handle(
        &self,
        note: &Note,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(text = %note.text, "reminding");
        // `processed_at` takes `WholeSeconds::now()`.
        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("reminders", "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::<Remind>("reminders"), |b| {
            let remind = subscriber("reminders", RemindHandler)
                .map_source(|_| InboxQueue::<Remind>::new("reminders"));
            b.include(remind.build());
        })
}

Example​

Example