bacnet-client 0.15.0

BACnet async client: TSM, segmentation, discovery, and service APIs
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

use super::*;
use std::sync::atomic::Ordering;

#[tokio::test]
async fn client_number_cancelled_stop_retains_task_joins_and_transport_cleanup() {
    let (mut client, inbound, mut outbound, gates) = harness(true).await;
    inject(&inbound, &number(77, 1), true).await;
    inject(&inbound, QUERY, false).await;
    bounded(gates.entered.acquire()).await.unwrap().forget();
    inject(&inbound, QUERY, true).await;
    gates.hold_stop.store(true, Ordering::SeqCst);
    {
        let stop = client.stop();
        tokio::pin!(stop);
        assert!(futures_util::poll!(&mut stop).is_pending());
    }
    assert!(
        client.network_number_task.is_some(),
        "canceled waiter retains Number join"
    );
    assert!(
        client.dispatch_task.is_some(),
        "canceled waiter retains dispatch join"
    );
    {
        let stop = client.stop();
        tokio::pin!(stop);
        bounded(async {
            tokio::select! {
                _ = &mut stop => panic!("transport stop must remain held"),
                permit = gates.stop_entered.acquire() => permit.unwrap().forget(),
            }
        })
        .await;
    }
    // Joining Number and dispatch released both network Arcs before transport stop.
    assert!(client.network_number_task.is_none());
    assert!(client.dispatch_task.is_none());
    bounded(gates.dropped.acquire()).await.unwrap().forget();
    gates.stop_release.add_permits(1);
    bounded(client.stop()).await.unwrap();
    bounded(inbound.closed()).await;
    gates.release.add_permits(2);
    drop(client);
    bounded(gates.transport_dropped.acquire())
        .await
        .unwrap()
        .forget();
    assert!(
        bounded(outbound.recv()).await.is_none(),
        "pending and queued Number sends canceled"
    );
}

#[tokio::test]
async fn client_number_drop_aborts_pending_and_queued_replies() {
    let (client, inbound, mut outbound, gates) = harness(true).await;
    inject(&inbound, &number(77, 1), true).await;
    inject(&inbound, QUERY, true).await;
    bounded(gates.entered.acquire()).await.unwrap().forget();
    inject(&inbound, QUERY, false).await;
    drop(client);
    bounded(gates.dropped.acquire()).await.unwrap().forget();
    bounded(gates.transport_dropped.acquire())
        .await
        .unwrap()
        .forget();
    bounded(inbound.closed()).await;
    gates.release.add_permits(2);
    assert!(bounded(outbound.recv()).await.is_none());
}

#[tokio::test]
async fn client_number_worker_queue_is_bounded_and_closes_on_abort() {
    use super::super::network_number;
    use bacnet_network::network_number::NumberControl;
    let (mut client, _, mut outbound, gates) = harness(true).await;
    let (tx, task) = network_number::spawn(&client.network);
    tx.try_send(NumberControl::NumberIs {
        number: 77,
        flag: 1,
    })
    .unwrap();
    tx.try_send(NumberControl::WhatIs).unwrap();
    bounded(gates.entered.acquire()).await.unwrap().forget();
    // The reply follows the announcement, so the layer already holds it.
    assert_eq!(client.network.local_network_number().get(), Some(77));
    for _ in 0..network_number::CAPACITY {
        tx.try_send(NumberControl::WhatIs).unwrap();
    }
    assert!(matches!(
        tx.try_send(NumberControl::WhatIs),
        Err(tokio::sync::mpsc::error::TrySendError::Full(_))
    ));
    task.abort();
    let _ = task.await;
    assert!(matches!(
        tx.try_send(NumberControl::WhatIs),
        Err(tokio::sync::mpsc::error::TrySendError::Closed(_))
    ));
    client.stop().await.unwrap();
    drop(client);
    assert!(bounded(outbound.recv()).await.is_none());
}