poll_interval

A subscription claims while its claims come back full, then waits the poll interval, one second unless the broker or the queue sets another.

InboxQueue::poll_interval sets the wait of one subscription, and SqlxBroker::poll_interval the wait of every subscription of the broker that sets none. A claim that found as many rows as it asked for is followed by the next one at once, so a backlog drains without a pause. A claim that found fewer waits the interval before the next one. A subscription by name takes the broker's interval. A publish through the broker into the subscription's table and group ends the wait at once, as Wake describes. So the interval bounds how long a row written by other means waits for its claim: a row another service inserts, a trigger writes, or the service writes with Insert. On Postgres listen_notify wakes a subscription for the rows of other processes too. A claim that fails is reported as an error of the subscription's stream, and the next claim waits one second. A shorter interval finds outside rows sooner, at the price of one more claim statement per interval and subscription.

Example

The warehouse writes exports into the table with its own SQL, so the queue claims five times a second; every other subscription of the broker waits five seconds.

Macros
Manual

examples/sqlx_subscriptions_poll.rs
use std::time::Duration;

use ruststream_sqlx::prelude::*;
use sqlx::sqlite::SqlitePoolOptions;
use sqlx::{Connection, SqliteConnection};

/// One row of `export_jobs`, which the warehouse's nightly job inserts with its own SQL.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "export_jobs")]
struct Export {
    #[field(id, generated)]
    id: i64,
    #[field(locked_until)]
    locked_until: Option<DateTime<Utc>>,
    #[field(payload)]
    payload: Vec<u8>,
}

/// The exports come from outside the service, so this queue claims five times a second.
#[subscriber(InboxQueue::<Export>::new("exports").poll_interval(Duration::from_millis(200)))]
async fn upload(file: &ExportFile) -> HandlerOutcome {
    tracing::info!(path = %file.path, "uploading an export");
    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());
    // Every other subscription of the broker waits five seconds after a claim that came back short.
    let broker = broker.poll_interval(Duration::from_secs(5));
    RustStream::new(AppInfo::new("exports", "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, |b| {
            b.include(upload);
        })
}

examples/manual/sqlx_subscriptions_poll.rs
use std::time::Duration;

use ruststream_sqlx::prelude::*;
use sqlx::sqlite::SqlitePoolOptions;
use sqlx::{Connection, SqliteConnection};

/// One row of `export_jobs`, which the warehouse's nightly job inserts with its own SQL.
#[derive(sqlx::FromRow)]
struct Export {
    id: i64,
    payload: Vec<u8>,
}

impl InboxTable for Export {
    type Id = i64;
    type Table = InboxSpec<(Lease<DateTime<Utc>>, Payload)>;
    const TABLE: Self::Table = InboxSpec::new("export_jobs", Column::new("id").generated())
        .lease(Column::new("locked_until"))
        .payload(Column::new("payload"));

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

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

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

struct Upload;

impl Handle<ExportFile> for Upload {
    fn handle(
        &self,
        file: &ExportFile,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(path = %file.path, "uploading an export");
        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());
    // Every other subscription of the broker waits five seconds after a claim that came back short.
    let broker = broker.poll_interval(Duration::from_secs(5));
    RustStream::new(AppInfo::new("exports", "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, |b| {
            // The exports come from outside the service, so this queue claims five times a second.
            let upload = subscriber("exports", Upload).map_source(|_| {
                InboxQueue::<Export>::new("exports").poll_interval(Duration::from_millis(200))
            });
            b.include(upload.build());
        })
}

Example​

Example