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

//! Context capture and bounded resource-loss delivery for target records.
use super::*;

impl<T: TransportPort + 'static> WriteAudit<'_, T> {
    pub(super) fn failure_ticket(
        &self,
        status: &Arc<AuditReporterStatus>,
        confirmed: bool,
        device: ObjectIdentifier,
        route: Option<Arc<ConfirmedRecipientRoute>>,
    ) -> Option<AuditFailureTicket<Arc<ConfirmedRecipientRoute>>> {
        let route = route?;
        let ticket = self
            .transactions
            .audit_failure_queue(status)?
            .observe(AuditFailureContext {
                epoch: status.configuration_epoch(),
                status: Arc::clone(status),
                device,
                confirmed,
                peer: route.canonical_peer.clone(),
                route,
                max_apdu: u32::from(self.config.device.max_apdu_length()),
            });
        if ticket.is_none() {
            // Checked admission identity exhaustion refuses capture without wrapping.
            status.complete_delivery(status.begin_delivery(), false);
        }
        ticket
    }

    pub(super) fn resource_drop(&self, pending: &PendingWrite) {
        record_resource_drop(
            self.transactions,
            self.network,
            &pending.status,
            pending.failure.clone(),
            pending
                .notification
                .target_timestamp
                .clone()
                .expect("completed record"),
        );
    }
}
pub(in crate::server) fn record_resource_drop<T: TransportPort + 'static>(
    transactions: &Arc<NotificationTransactions>,
    network: &Arc<NetworkLayer<T>>,
    status: &Arc<AuditReporterStatus>,
    ticket: Option<AuditFailureTicket<Arc<ConfirmedRecipientRoute>>>,
    timestamp: bacnet_types::primitives::BACnetTimeStamp,
) {
    let Some(ticket) = ticket else {
        return;
    };
    let Some(queue) = transactions.audit_failure_queue(status) else {
        return;
    };
    let Some(mut worker) = queue.record_drop(transactions, ticket, timestamp, 1) else {
        return;
    };
    let network = Arc::clone(network);
    transactions.spawn(async move {
        while let Some((batch, _permit, reserved)) = worker.next().await {
            let completion = DeliveryCompletion::auditing_failure(
                Arc::clone(&batch.context.status),
                batch.context.epoch,
            );
            let deadline = tokio::time::Instant::now() + DELIVERY_TIMEOUT;
            let invoke = reserved
                .as_ref()
                .map_or(0, |(operation, _)| operation.invoke_id());
            let Some(bytes) = encode_notification(
                &batch.notification(),
                batch.context.confirmed,
                batch.context.max_apdu,
                invoke,
            ) else {
                continue;
            };
            let delivered =
                deliver(&network, &batch.context.route, &bytes, reserved, deadline).await;
            if let Some(completion) = completion {
                completion.finish(delivered);
            }
        }
    });
}