bacnet-server 0.15.0

BACnet server: APDU dispatch and service handlers
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
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62

//! The work adding and removing objects leaves for the server (#1341,
//! #1440).
//!
//! `ObjectDatabase::add` and `remove` judge again the Pulse Converters whose
//! Input_Reference names the object, under the caller's guard, and queue
//! those whose Reliability changed; `add` also queues each Schedule holding
//! a refused reference to the new object. The rest is the server's: the
//! Schedule's retry to the new object ([`crate::schedule::retry_for_created`])
//! and the COV for what changed. A CreateObject or DeleteObject takes the
//! queue under its own guard and adds the fanout, and any Command run a
//! retry starts, to the request's, and an application's own `add` or
//! `remove` wakes the waker [`install_waker`] puts on the database, which
//! the server's Schedule task waits on to take the queue under the next
//! guard it gets ([`settle_committed`]). Lock order is the server's: the
//! database guard, then the COV table.

use std::sync::Arc;

use bacnet_objects::database::ObjectDatabase;
use tokio::sync::{Notify, RwLock};

use crate::committed_cov::{BackgroundCommit, CommittedCov};
use crate::cov::CovSubscriptionTable;

/// Take the work queued on `db_w`, the guard on `db`, and return the COV
/// fanout it owes, to send once that guard is dropped.
pub(crate) async fn settle(
    db: &Arc<RwLock<ObjectDatabase>>,
    db_w: &mut ObjectDatabase,
    cov_table: &RwLock<CovSubscriptionTable>,
) -> CommittedCov {
    let work = db_w.take_membership_work_internal();
    if work.is_empty() {
        return CommittedCov::default();
    }
    let mut commit = BackgroundCommit::new();
    for oid in work.changed {
        commit.changed(oid);
    }
    crate::schedule::retry_for_created(db_w, &mut commit, &work.schedule_retries);
    commit.finish(db, db_w, cov_table).await
}

/// [`settle`] under a guard of its own, for the server's background task.
pub(crate) async fn settle_committed(
    db: &Arc<RwLock<ObjectDatabase>>,
    cov_table: &RwLock<CovSubscriptionTable>,
) -> CommittedCov {
    let mut db_w = db.write().await;
    settle(db, &mut db_w, cov_table).await
}

/// Install a waker on `db` that wakes the returned notify whenever a change
/// of membership queues work, armed once so work queued before the server
/// started is taken too.
pub(crate) fn install_waker(db: &mut ObjectDatabase) -> Arc<Notify> {
    let notify = Arc::new(Notify::new());
    let waker = Arc::clone(&notify);
    db.set_membership_waker_internal(Some(Arc::new(move || waker.notify_one())));
    notify.notify_one();
    notify
}