This enqueues a multicast-group queue item for the given gateways and returns the frame-counter of the multicast downlink. This function locks the multicast-group to avoid race-conditions with scheduling time and frame-counters.
(
qi: MulticastGroupQueueItem,
gateway_ids: &[EUI64],
)
| 439 | // This function locks the multicast-group to avoid race-conditions with scheduling time and |
| 440 | // frame-counters. |
| 441 | pub async fn enqueue( |
| 442 | qi: MulticastGroupQueueItem, |
| 443 | gateway_ids: &[EUI64], |
| 444 | ) -> Result<(Vec<Uuid>, u32), Error> { |
| 445 | qi.validate()?; |
| 446 | let mut c = get_async_db_conn().await?; |
| 447 | let conf = config::get(); |
| 448 | let (ids, f_cnt) = c |
| 449 | .transaction::<(Vec<Uuid>, u32), Error, _>(async |c| { |
| 450 | let mut ids: Vec<Uuid> = Vec::new(); |
| 451 | let query = multicast_group::dsl::multicast_group.find(&qi.multicast_group_id); |
| 452 | #[cfg(feature = "postgres")] |
| 453 | let query = query.for_update(); |
| 454 | let mg: MulticastGroup = query |
| 455 | .get_result(c) |
| 456 | .await |
| 457 | .map_err(|e| Error::from_diesel(e, qi.multicast_group_id.to_string()))?; |
| 458 | |
| 459 | match mg.group_type.as_ref() { |
| 460 | "B" => { |
| 461 | // get ping nb |
| 462 | let ping_nb = 1 << (7 - mg.class_b_ping_slot_periodicity) as usize; |
| 463 | |
| 464 | // get max. gps epoch time. |
| 465 | let res: Option<i64> = |
| 466 | multicast_group_queue_item::dsl::multicast_group_queue_item |
| 467 | .select(dsl::max( |
| 468 | multicast_group_queue_item::dsl::emit_at_time_since_gps_epoch, |
| 469 | )) |
| 470 | .filter( |
| 471 | multicast_group_queue_item::dsl::multicast_group_id |
| 472 | .eq(&qi.multicast_group_id), |
| 473 | ) |
| 474 | .first(c) |
| 475 | .await?; |
| 476 | |
| 477 | // Get timestamp after which we must generate the next ping-slot. |
| 478 | let ping_slot_after_gps_time = match res { |
| 479 | Some(v) => Duration::try_milliseconds(v).unwrap_or_default(), |
| 480 | None => (Utc::now() |
| 481 | + Duration::from_std(conf.network.scheduler.multicast_class_b_margin) |
| 482 | .unwrap()) |
| 483 | .to_gps_time(), |
| 484 | }; |
| 485 | |
| 486 | let emit_at_time_since_gps_epoch = classb::get_next_ping_slot_after( |
| 487 | ping_slot_after_gps_time, |
| 488 | &mg.mc_addr, |
| 489 | ping_nb, |
| 490 | )?; |
| 491 | |
| 492 | let scheduler_run_after_ts = emit_at_time_since_gps_epoch.to_date_time() |
| 493 | - Duration::from_std(2 * conf.network.scheduler.interval).unwrap(); |
| 494 | |
| 495 | for gateway_id in gateway_ids { |
| 496 | let qi = MulticastGroupQueueItem { |
| 497 | scheduler_run_after: scheduler_run_after_ts, |
| 498 | multicast_group_id: mg.id, |