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
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101

use super::event_notifications::resolve_committed_event_enrollment_transition;
use super::*;

/// Owned handles and settings the event-enrollment task moves into its loop.
pub(super) struct EventEnrollmentTask<T: TransportPort + 'static> {
    pub(super) db: Arc<RwLock<ObjectDatabase>>,
    pub(super) network: Arc<NetworkLayer<T>>,
    pub(super) comm_state: Arc<CommState>,
    pub(super) learned_routers: Arc<Mutex<LearnedRouterCache>>,
    pub(super) notification_transactions: Arc<NotificationTransactions>,
    pub(super) device_bindings: Arc<RwLock<DeviceBindingTable>>,
    pub(super) suppressions: Arc<super::event_suppression::EventSuppressions>,
    /// Evaluation period, already clamped by the caller.
    pub(super) period: Duration,
    pub(super) retry_ms: u64,
    pub(super) local_apdu_capacity: u32,
}

pub(super) fn spawn_event_enrollment_task<T: TransportPort + 'static>(
    task: EventEnrollmentTask<T>,
) -> JoinHandle<()> {
    let EventEnrollmentTask {
        db,
        network,
        comm_state,
        learned_routers,
        notification_transactions,
        device_bindings,
        suppressions,
        period,
        retry_ms,
        local_apdu_capacity,
    } = task;
    let evaluation_interval_secs = period.as_secs().max(1);
    let owner = notification_transactions.audit_owner_lease();
    super::heap_futures::spawn_boxed(move || async move {
        let _owner = owner;
        let mut interval = tokio::time::interval(period);
        // A stalled runtime must not fire a burst of catch-up passes; the
        // adjacent intrinsic-reporting task sets this for the same reason.
        interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
        loop {
            interval.tick().await;
            // Evaluate, commit, and project every result in enrollment order
            // while one database guard is held. The projector consumes
            // committed timestamp/ACK/clock inputs only; it neither stages
            // another timestamp nor repeats transition actions. The guard is
            // dropped before the first send.
            let (report, outbound) = {
                let mut db_guard = db.write().await;
                let evaluation = crate::event_enrollment::evaluate_event_enrollments_for_delivery(
                    &mut db_guard,
                    evaluation_interval_secs,
                );
                let outbound = evaluation
                    .deliveries
                    .into_iter()
                    .filter_map(|committed| {
                        resolve_committed_event_enrollment_transition(&db_guard, committed)
                    })
                    .filter(|(_, distribute, _)| *distribute)
                    .map(|(oid, _, transition)| (oid, transition))
                    .collect::<Vec<_>>();
                (evaluation.report, outbound)
            };
            for transition in &report.transitions {
                debug!(
                    enrollment = %transition.enrollment_oid,
                    monitored = %transition.monitored_oid,
                    from = ?transition.change.from,
                    to = ?transition.change.to,
                    distribute = transition.distribute,
                    "Event enrollment: state changed"
                );
            }
            crate::event_enrollment::log_evaluation_report(&report);
            for (oid, transition) in outbound {
                BACnetServer::<T>::build_and_send_event_notification_with_bindings(
                    &EventDelivery {
                        db: &db,
                        network: &network,
                        comm_state: &comm_state,
                        learned_routers: &learned_routers,
                        notification_transactions: &notification_transactions,
                        device_bindings: &device_bindings,
                        suppressions: &suppressions,
                        retry_timeout_ms: retry_ms,
                        local_apdu_capacity,
                    },
                    &oid,
                    transition,
                )
                .await;
            }
        }
    })
}

#[cfg(test)]
#[path = "event_enrollment_notification_tests.rs"]
mod tests;