archimedes 0.4.0

High performance Rust/PostgreSQL job queue (also suitable for getting jobs generated by PostgreSQL triggers/functions out into a different work queue)
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

use std::time::Duration;

use archimedes_shutdown_signal::ShutdownSignal;
use futures::{stream, Stream};
use sqlx::{postgres::PgListener, PgPool};

use crate::errors::Result;

#[derive(Debug)]
pub enum StreamSource {
    Polling,
    PgListener,
}

/// Returns a stream that yield on postgres `NOTIFY 'jobs:insert'` and on interval specified by the
/// poll_interval argument
pub async fn job_signal_stream(
    pg_pool: PgPool,
    poll_interval: Duration,
    shutdown_signal: ShutdownSignal,
) -> Result<impl Stream<Item = StreamSource>> {
    let interval = tokio::time::interval(poll_interval);

    let mut pg_listener = PgListener::connect_with(&pg_pool).await?;
    pg_listener.listen("jobs:insert").await?;

    let stream = stream::unfold((interval, pg_listener, shutdown_signal), |mut f| async {
        tokio::select! {
            _ = (f.0).tick() => Some((StreamSource::Polling, f)),
            _ = (f.1).recv() => Some((StreamSource::PgListener, f)),
            _ = &mut f.2 => None,
        }
    });

    Ok(stream)
}