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
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125

use super::super::*;
use bacnet_endpoint_core::coordinator::CanonicalPeer;
#[path = "../../tests/support/sc_number_socket.rs"]
mod socket;
use socket::*;

#[tokio::test]
async fn sc_number_server_blocked_write_flood_ack_handler_and_canceled_stop() {
    let (transport, peer, observed) = fixture();
    let (server, ()) = bounded(async {
        tokio::join!(
            BACnetServer::start(
                ServerConfig {
                    device: bacnet_objects::DeviceIdentity::new(7, 0).unwrap(),
                    ..Default::default()
                },
                database(),
                transport
            ),
            peer.accept()
        )
    })
    .await;
    let mut server = server.unwrap();
    let (operation, acknowledged) = server
        .notification_transactions
        .reserve(
            CanonicalPeer::direct(&PEER),
            ConfirmedServiceChoice::CONFIRMED_AUDIT_NOTIFICATION,
        )
        .unwrap();
    peer.learn_and_query().await;
    count(&observed.numbers_started, 1).await;
    // The worker cannot drain its 256-entry input while physically sending.
    // Pace by actual network intake, not by sleeps or the SC origin quota.
    for sequence in 3..=260 {
        peer.send(false, &[1, 0x80, 0x12]).await;
        bounded(async {
            while server.test_network().network_control_ingress_sequence() < sequence {
                tokio::task::yield_now().await;
            }
        })
        .await;
    }
    peer.send(
        false,
        &[
            1,
            0,
            0x20,
            operation.invoke_id(),
            ConfirmedServiceChoice::CONFIRMED_AUDIT_NOTIFICATION.to_raw(),
        ],
    )
    .await;
    assert!(matches!(
        bounded(acknowledged).await.unwrap(),
        CovAckResult::Ack
    ));
    drop(operation);
    peer.send(false, &write_npdu()).await;
    handled(server.database()).await;
    assert_eq!(observed.numbers_completed.load(Ordering::SeqCst), 0);
    observed.hold_disconnect.store(true, Ordering::SeqCst);
    {
        let stop = server.stop();
        tokio::pin!(stop);
        bounded(async {
            tokio::select! { biased;
                _ = &mut stop => panic!("held transport stop completed"),
                _ = count(&observed.disconnect_started, 1) => {}
            }
        })
        .await;
    }
    assert_eq!(observed.numbers_dropped.load(Ordering::SeqCst), 1);
    observed.disconnect_release.add_permits(1);
    bounded(server.stop()).await.unwrap();
    count(&observed.socket_dropped, 1).await;
    assert_eq!(observed.numbers_started.load(Ordering::SeqCst), 1);
    assert_eq!(observed.numbers_completed.load(Ordering::SeqCst), 0);
    assert_eq!(
        tokio::runtime::Handle::current()
            .metrics()
            .num_alive_tasks(),
        0
    );
}

#[tokio::test]
async fn sc_number_server_live_release_and_bare_drop_join() {
    for bare_drop in [false, true] {
        let (transport, peer, observed) = fixture();
        let (server, ()) = bounded(async {
            tokio::join!(
                BACnetServer::start(
                    ServerConfig {
                        device: bacnet_objects::DeviceIdentity::new(7, 0).unwrap(),
                        ..Default::default()
                    },
                    database(),
                    transport
                ),
                peer.accept()
            )
        })
        .await;
        let mut server = server.unwrap();
        peer.learn_and_query().await;
        count(&observed.numbers_started, 1).await;
        if !bare_drop {
            observed.number_release.add_permits(1);
            peer.number().await;
            assert_eq!(observed.numbers_completed.load(Ordering::SeqCst), 1);
            bounded(server.stop()).await.unwrap();
        }
        drop(server);
        count(&observed.socket_dropped, 1).await;
        assert_eq!(observed.numbers_dropped.load(Ordering::SeqCst), 1);
        assert_eq!(
            observed.numbers_completed.load(Ordering::SeqCst),
            usize::from(!bare_drop)
        );
    }
}