Перейти к основному содержимому
Эта страница ещё не переведена

Ниже показан английский оригинал.

Wake

A publish through the broker wakes the subscriptions of its table and group in the same process at once; a row written by other means waits the poll.

A subscription claims rows as long as its claims come back full. After a claim that found fewer rows than it asked for, it waits the poll interval: one second unless the broker or the subscription sets another. Two things end the wait early: a publish from the same process, and a Postgres notification.

Every publish through the broker wakes: a Repository, a route, a reply or an Out slot published through either of them, and a test's input through the harness. In a table with groups, a publish wakes the subscriptions of its own group, and the others keep their wait. In a table without groups, it wakes every subscription of the table. A subscription busy with a claim keeps the wake-up, so its next wait returns at once.

A row the service writes itself wakes nobody: an insert, a statement of the service's own, a write through a handler's Tx. Such a row waits the poll interval, or a notification the service sends.

The wake-up costs a publish one atomic operation per subscription it wakes. A subscription pays nothing per message for it. A long poll interval therefore costs a service that publishes its own work no latency: its subscriptions poll only for rows other programs write.

Example​

The broker polls an empty queue once an hour. The welcome a signup's handler replies with still reaches its subscription at once, because the reply's row wakes it.

examples/sqlx_wake.rs
use ruststream::OutgoingMessage;
use sqlx::{Sqlite, SqliteConnection};

/// One row of `jobs`: the group column holds the name a message was published to.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "jobs")]
struct Job {
    #[field(id, generated)]
    id: i64,
    #[field(group)]
    name: String,
    #[field(locked_until)]
    locked_until: Option<DateTime<Utc>>,
    #[field(payload)]
    payload: Vec<u8>,
}

/// The service's own insert: the name goes to the group column, the bytes to the payload.
impl Publish<Sqlite> for Job {
    async fn publish(
        conn: &mut SqliteConnection,
        message: &OutgoingMessage<'_>,
    ) -> Result<(), sqlx::Error> {
        sqlx::query("INSERT INTO jobs (name, payload) VALUES (?, ?)")
            .bind(message.name())
            .bind(message.payload())
            .execute(conn)
            .await?;
        Ok(())
    }
}

#[subscriber(InboxQueue::<Job>::new("signups"), reply)]
async fn greet(signup: &Signup) -> Welcome {
    Welcome {
        email: signup.email.clone(),
    }
}

// The welcome's row wakes this subscription: no poll interval in between.
#[subscriber(InboxQueue::<Job>::new("welcome"))]
async fn send(welcome: &Welcome) -> HandlerOutcome {
    tracing::info!(to = %welcome.email, "sending the welcome");
    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());
    // The hourly poll reaches a row the service wrote with its own SQL.
    let broker = broker
        .poll_interval(Duration::from_secs(3600))
        .route::<Job>("signups");
    RustStream::new(AppInfo::new("signup", "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(greet).out_reply(Repository::<Job>::default());
            b.include(send);
        })
}