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
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
use super::*;
impl<T: TransportPort + 'static> BACnetServer<T> {
/// Dispatch a received APDU.
///
/// ConfirmedRequest and UnconfirmedRequest handlers are spawned as
/// independent tasks so the dispatch loop can immediately process the
/// next incoming APDU. This allows the server to handle multiple
/// client requests concurrently (e.g. concurrent ReadProperty via
/// the RwLock on ObjectDatabase).
///
/// Fast-path APDU types (SimpleAck, Error, Reject, Abort, SegmentAck)
/// remain inline since they are sub-microsecond TSM lookups.
#[allow(clippy::too_many_arguments)]
pub(super) async fn dispatch(
db: &Arc<RwLock<ObjectDatabase>>,
network: &Arc<NetworkLayer<T>>,
cov_table: &Arc<RwLock<CovSubscriptionTable>>,
seg_ack_senders: &Arc<Mutex<HashMap<SegKey, mpsc::Sender<SegmentAckPdu>>>>,
cov_in_flight: &Arc<Semaphore>,
server_tsm: &Arc<Mutex<ServerTsm>>,
comm_state: &Arc<AtomicU8>,
dcc_timer: &Arc<Mutex<Option<JoinHandle<()>>>>,
config: &Arc<ServerConfig>,
source_mac: &[u8],
apdu: Apdu,
mut received: bacnet_network::layer::ReceivedApdu,
) {
match apdu {
Apdu::ConfirmedRequest(req) => {
let reply_tx = received.reply_tx.take();
let db = Arc::clone(db);
let network = Arc::clone(network);
let cov_table = Arc::clone(cov_table);
let seg_ack_senders = Arc::clone(seg_ack_senders);
let cov_in_flight = Arc::clone(cov_in_flight);
let server_tsm = Arc::clone(server_tsm);
let comm_state = Arc::clone(comm_state);
let dcc_timer = Arc::clone(dcc_timer);
let config = Arc::clone(config);
let source_mac = MacAddr::from_slice(source_mac);
let source_network = received.source_network.clone();
tokio::spawn(async move {
Self::handle_confirmed_request(
&db,
&network,
&cov_table,
&seg_ack_senders,
&cov_in_flight,
&server_tsm,
&comm_state,
&dcc_timer,
&config,
&source_mac,
source_network,
req,
reply_tx,
)
.await;
});
}
Apdu::UnconfirmedRequest(req) => {
let db = Arc::clone(db);
let network = Arc::clone(network);
let config = Arc::clone(config);
let comm_state = Arc::clone(comm_state);
tokio::spawn(async move {
Self::handle_unconfirmed_request(
&db,
&network,
&config,
&comm_state,
req,
&received,
)
.await;
});
}
// Fast paths — remain inline (sub-microsecond TSM lookups)
Apdu::SimpleAck(sa) => {
let mut tsm = server_tsm.lock().await;
let peer = MacAddr::from_slice(source_mac);
if !tsm.record_result(
&peer,
received.source_network.as_ref(),
sa.invoke_id,
CovAckResult::Ack,
) {
tsm.record_result(&MacAddr::new(), None, sa.invoke_id, CovAckResult::Ack);
}
debug!(
invoke_id = sa.invoke_id,
"SimpleAck received for outgoing confirmed notification"
);
}
Apdu::Error(err) => {
let mut tsm = server_tsm.lock().await;
let peer = MacAddr::from_slice(source_mac);
if !tsm.record_result(
&peer,
received.source_network.as_ref(),
err.invoke_id,
CovAckResult::Error,
) {
tsm.record_result(&MacAddr::new(), None, err.invoke_id, CovAckResult::Error);
}
debug!(
invoke_id = err.invoke_id,
error_class = err.error_class.to_raw(),
error_code = err.error_code.to_raw(),
"Error received for outgoing confirmed notification"
);
}
Apdu::Reject(rej) => {
let mut tsm = server_tsm.lock().await;
let peer = MacAddr::from_slice(source_mac);
if !tsm.record_result(
&peer,
received.source_network.as_ref(),
rej.invoke_id,
CovAckResult::Error,
) {
tsm.record_result(&MacAddr::new(), None, rej.invoke_id, CovAckResult::Error);
}
debug!(
invoke_id = rej.invoke_id,
"Reject received for outgoing confirmed notification"
);
}
Apdu::Abort(abort) if !abort.sent_by_server => {
let mut tsm = server_tsm.lock().await;
let peer = MacAddr::from_slice(source_mac);
if !tsm.record_result(
&peer,
received.source_network.as_ref(),
abort.invoke_id,
CovAckResult::Error,
) {
tsm.record_result(&MacAddr::new(), None, abort.invoke_id, CovAckResult::Error);
}
debug!(
invoke_id = abort.invoke_id,
"Abort received for outgoing confirmed notification"
);
}
Apdu::SegmentAck(sa) => {
let key = (
MacAddr::from_slice(source_mac),
received.source_network.clone(),
sa.invoke_id,
);
let senders = seg_ack_senders.lock().await;
if let Some(tx) = senders.get(&key) {
let _ = tx.try_send(sa);
} else {
debug!(
invoke_id = sa.invoke_id,
"Server ignoring SegmentAck for unknown transaction"
);
}
}
_ => {
debug!("Server ignoring unhandled APDU type");
}
}
}
}