bacnet-network 0.12.0

BACnet network layer: routing, router tables, priority channels
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
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150

//! The router's shared local APDU queue: what a full queue drops, what an
//! admitted APDU keeps, and the raw receiver's policy. The modules below cover
//! the receiver's lifecycle, concurrent producers and the per-source quota.
//! All of them run a real router through
//! [`ingress_harness`](super::ingress_harness).

use super::ingress_harness::*;
use super::*;
use crate::layer::QueueAdmissionSnapshot;
use bacnet_transport::port::ReceivedNpdu;
use tokio::sync::oneshot;
use tokio::time::timeout;

#[tokio::test(start_paused = true)]
async fn full_shared_local_queue_drops_arrivals_in_all_four_branches_and_keeps_forwarding() {
    for branch in BRANCHES {
        let (ports, mut peers) = fixture(2);
        let (mut router, mut apdus) = launch(ports, RouterOptions::new().track_admission()).await;
        let counters = apdus.counters();
        drain_announcements(&mut peers).await;
        let mut accepted_reply = fill(&mut peers).await;
        for port in 0..2 {
            dropped_arrival(&mut peers, branch, port).await;
            assert_eq!(
                counters.snapshot(),
                QueueAdmissionSnapshot {
                    current_depth: 256,
                    high_water: 256,
                    full_drops: port as u64 + 1,
                    fairness_drops: 0,
                    closed_drops: 0,
                }
            );
        }
        assert_eq!(
            accepted_reply.try_recv(),
            Err(oneshot::error::TryRecvError::Empty)
        );
        for id in 0..256 {
            let apdu = apdus.try_recv().unwrap();
            assert_apdu(&apdu, LocalBranch::NoDnet, usize::from(id / 128), id);
            if id == 0 {
                apdu.reply_tx
                    .unwrap()
                    .send(Bytes::from_static(b"reply"))
                    .unwrap();
            }
            assert_eq!(counters.snapshot().current_depth, 255 - usize::from(id));
        }
        assert_eq!(accepted_reply.await.unwrap(), Bytes::from_static(b"reply"));
        assert!(matches!(
            apdus.try_recv(),
            Err(mpsc::error::TryRecvError::Empty)
        ));
        peers[0].tx.send(incoming(None, 257)).await.unwrap();
        let apdu = timeout(Duration::from_secs(2), apdus.recv())
            .await
            .unwrap()
            .unwrap();
        assert_apdu(&apdu, LocalBranch::NoDnet, 0, 257);
        assert_eq!(
            counters.snapshot(),
            QueueAdmissionSnapshot {
                high_water: 256,
                full_drops: 2,
                ..Default::default()
            }
        );
        router.stop().await;
    }
}

#[tokio::test(start_paused = true)]
async fn admitted_local_branches_preserve_metadata_and_reply_ownership() {
    let (ports, mut peers) = fixture(2);
    let (mut router, mut apdus) = launch(ports, RouterOptions::new().track_admission()).await;
    drain_announcements(&mut peers).await;
    for branch in BRANCHES {
        let (reply_tx, mut reply_rx) = oneshot::channel();
        let mut arrival = local(branch, 0, 42);
        arrival.reply_tx = Some(reply_tx);
        peers[0].tx.send(arrival).await.unwrap();
        branch_forward(&mut peers, branch, 0, 42).await;
        barrier(&mut peers[0]).await;
        assert_eq!(apdus.counters().snapshot().current_depth, 1);
        let apdu = apdus.recv().await.unwrap();
        assert_apdu(&apdu, branch, 0, 42);
        assert_eq!(apdu.clone().ingress_network, apdu.ingress_network);
        if matches!(branch, LocalBranch::RemoteBroadcast) {
            assert!(apdu.reply_tx.is_none());
            assert!(reply_rx.await.is_err());
        } else {
            assert_eq!(
                reply_rx.try_recv(),
                Err(oneshot::error::TryRecvError::Empty)
            );
            apdu.reply_tx
                .unwrap()
                .send(Bytes::from_static(b"reply"))
                .unwrap();
            assert_eq!(reply_rx.await.unwrap(), Bytes::from_static(b"reply"));
        }
        assert_eq!(
            apdus.counters().snapshot(),
            QueueAdmissionSnapshot {
                high_water: 1,
                ..Default::default()
            }
        );
    }
    assert_quiet(&mut peers);
    router.stop().await;
}

#[tokio::test(start_paused = true)]
async fn legacy_local_receiver_retains_type_and_nonblocking_full_closed_policy() {
    let (ports, mut peers) = fixture(2);
    let (mut router, mut apdus): (_, mpsc::Receiver<ReceivedApdu>) =
        launch(ports, RouterOptions::new()).await;
    drain_announcements(&mut peers).await;
    let accepted_reply = fill(&mut peers).await;
    for branch in BRANCHES {
        dropped_arrival(&mut peers, branch, 0).await;
    }
    assert_eq!(apdus.len(), 256);
    for id in 0..256 {
        assert_apdu(
            &apdus.try_recv().unwrap(),
            LocalBranch::NoDnet,
            usize::from(id / 128),
            id,
        );
    }
    assert!(accepted_reply.await.is_err());
    apdus.close();
    for branch in BRANCHES {
        dropped_arrival(&mut peers, branch, 1).await;
    }
    assert!(apdus.recv().await.is_none());
    router.stop().await;
}

#[path = "admission_lifecycle_tests.rs"]
mod lifecycle_tests;

#[path = "admission_concurrency_tests.rs"]
mod concurrency_tests;

#[path = "fairness_tests.rs"]
mod fairness_tests;