Function get_schedulable_queue_items

chirpstack/src/storage/multicast.rs:661–722

(limit: usize)

}

pub async fn get_schedulable_queue_items(limit: usize) -> Result<Vec<MulticastGroupQueueItem>> {
    let mut c = get_async_db_conn().await?;
    c.transaction::<Vec<MulticastGroupQueueItem>, Error, _>(async |c| {
                let conf = config::get();
                diesel::sql_query(if cfg!(feature = "sqlite") {
                    r#"
                        update
                            multicast_group_queue_item
                        set
                            scheduler_run_after = ?3
                        where
                            id in (
                                select
                                    id
                                from
                                    multicast_group_queue_item
                                where
                                    scheduler_run_after <= ?2
                                order by
                                    created_at
                                limit ?1
                            )
                        returning *
                    "#
                } else {
                    r#"
                        update
                            multicast_group_queue_item
                        set
                            scheduler_run_after = $3
                        where
                            id in (
                                select
                                    qi.id
                                from
                                    multicast_group_queue_item qi
                                inner join gateway g
                                    on g.gateway_id = qi.gateway_id
                                where
                                    qi.scheduler_run_after <= $2
                                    -- check that the gateway is online, except when the item already has expired
                                    and ($2 - make_interval(secs => g.stats_interval_secs * 2) <= g.last_seen_at or expires_at <= $2)
                                order by
                                    qi.created_at
                                limit $1
                                for update skip locked
                            )
                        returning *
                    "#
                })
                .bind::<diesel::sql_types::Integer, _>(limit as i32)
                .bind::<fields::sql_types::Timestamptz, _>(Utc::now())
                .bind::<fields::sql_types::Timestamptz, _>(
                    Utc::now() + Duration::from_std(conf.network.scheduler.scheduler_lock_duration).unwrap(),
                )
                .load(c)
                .await
                .map_err(|e| Error::from_diesel(e, "".into()))