mod actions;
pub mod control_policy;
mod counters;
mod dispatch;
mod engine;
mod hosted_host;
mod limits;
mod local_control;
mod local_delivery;
mod options;
mod originate;
mod port_groups;
mod replies;
mod send_queue;
mod send_warning;
mod sender;
pub use replies::RouterResponse;
use std::ops::{Deref, DerefMut};
use std::sync::atomic::AtomicU64;
use std::sync::Arc;
use bacnet_npdu::Mac;
pub use bacnet_router_core::router::{
Congestion, Duplicates, OriginateError, Request, SourceSelection,
};
use bacnet_router_core::{Instant, PortId};
use bacnet_transport::port::{DataAttribute, TransportPort};
use bacnet_types::error::Error;
use bacnet_types::MacAddr;
use bytes::Bytes;
pub use counters::Counters;
pub use originate::OriginateFailure;
use tokio::sync::mpsc;
use tokio::task::JoinHandle;
use crate::layer::ReceivedNetworkControl;
use crate::layer::{AdmissionReceiver, AdmissionSender, QueueAdmissionCounters, ReceivedApdu};
pub use control_policy::{ControlAuthContext, ControlAuthorizer, ControlClass};
pub use control_policy::{ControlDecisionCounters, ControlGate, ControlPolicy};
pub use control_policy::{ControlServiceCounters, ControlTrust};
use dispatch::{Outputs, PortDispatch, Shared};
use engine::{tokio_clock, Engine, EngineGuard, PORTS};
pub use options::{LocalApduReceiver, NetworkControlReceiver, RouterOptions, StartedRouter};
use port_groups::PortGroups;
#[cfg(test)]
use {
bacnet_encoding::npdu::{encode_npdu, Npdu},
bacnet_npdu::NetworkMessageType,
bacnet_transport::port::TransportProvenance,
bytes::BytesMut,
std::time::Duration,
};
#[derive(Debug)]
enum SendRequest {
Unicast {
npdu: Bytes,
mac: MacAddr,
data_attributes: Vec<DataAttribute>,
reject_origin: Option<bacnet_router_core::router::RejectOrigin>,
response: Option<replies::ReplyCompletion>,
},
Broadcast {
npdu: Bytes,
data_attributes: Vec<DataAttribute>,
reject_origin: Option<bacnet_router_core::router::RejectOrigin>,
},
}
pub struct RouterPort<T: TransportPort> {
pub transport: T,
pub network_number: u16,
}
pub struct BACnetRouter {
replies: Arc<replies::ReplyLifetime>,
outputs: std::sync::Weak<Outputs>,
shared: Arc<Shared>,
dispatch_tasks: Vec<JoinHandle<()>>,
sender_tasks: Vec<JoinHandle<()>>,
aging_task: Option<JoinHandle<()>>,
limit_tasks: Vec<JoinHandle<()>>,
network_control_ingress_sequence: Arc<AtomicU64>,
}
pub struct RouterTableGuard<'a> {
router: EngineGuard<'a>,
}
impl Deref for RouterTableGuard<'_> {
type Target = crate::router_table::RouterTable;
fn deref(&self) -> &Self::Target {
self.router.table()
}
}
impl DerefMut for RouterTableGuard<'_> {
fn deref_mut(&mut self) -> &mut Self::Target {
self.router.table_mut()
}
}
impl std::fmt::Debug for RouterTableGuard<'_> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
self.router.table().fmt(f)
}
}
impl BACnetRouter {
pub async fn start<T, A, C>(
ports: Vec<RouterPort<T>>,
options: RouterOptions<A, C>,
) -> Result<StartedRouter<A, C>, Error>
where
T: TransportPort + 'static,
A: LocalApduReceiver,
C: NetworkControlReceiver,
{
let (control_tx, network_control) = if options.wants_network_control() {
let (tx, rx, counters) = AdmissionReceiver::channel(C::TRACKED);
(Some(tx), Some(C::from_queue(rx, counters)))
} else {
(None, None)
};
let gate = Arc::new(options.control_gate());
let (router, rx, counters) = Self::start_dispatch(
ports,
A::TRACKED,
gate,
control_tx,
options.engine_options(),
)
.await?;
Ok(StartedRouter {
router,
apdus: A::from_queue(rx, counters),
network_control,
})
}
pub fn control_snapshot(&self) -> control_policy::ControlDecisionCounters {
self.shared.gate.snapshot()
}
pub fn counters(&self) -> Counters {
self.shared.engine.counters()
}
pub fn address_length_drops(&self) -> u64 {
self.counters().address_length_drops
}
pub fn global_broadcast_dadr_drops(&self) -> u64 {
self.counters().global_broadcast_dadr_drops
}
pub fn broadcast_pdu_type_drops(&self) -> u64 {
self.counters().broadcast_pdu_type_drops
}
pub fn group_dadr_drops(&self) -> u64 {
self.counters().group_dadr_drops
}
async fn start_dispatch<T: TransportPort + 'static>(
mut ports: Vec<RouterPort<T>>,
track_depth: bool,
control: Arc<control_policy::ControlGate>,
network_control: Option<AdmissionSender<ReceivedNetworkControl>>,
engine_options: bacnet_router_core::router::Options,
) -> Result<(Self, mpsc::Receiver<ReceivedApdu>, QueueAdmissionCounters), Error> {
if ports.is_empty() || ports.len() > PORTS {
return Err(Error::Encoding(format!(
"A router needs 1 to {PORTS} ports, not {}",
ports.len()
)));
}
{
let mut seen = std::collections::HashSet::new();
for port in &ports {
if matches!(port.network_number, 0 | 0xFFFF) {
return Err(Error::Encoding(format!(
"A router port can't be on network {}: ports take 1 to 65534",
port.network_number
)));
}
if !seen.insert(port.network_number) {
return Err(Error::Encoding(format!(
"Duplicate network number {} in router ports",
port.network_number
)));
}
}
}
let (local_tx, local_rx, counters) = AdmissionReceiver::channel(track_depth);
let mut port_receivers = Vec::new();
let mut configs = Vec::new();
let mut networks = Vec::new();
let mut limit_changes = Vec::new();
for port in &mut ports {
port.transport.prepare_for_routing();
let rx = port.transport.start().await?;
port_receivers.push(rx);
let mac = Mac::from_slice(port.transport.local_mac()).map_err(|_| {
Error::Encoding(format!(
"Router port on network {} has a MAC longer than {} octets",
port.network_number,
bacnet_npdu::MAX_MAC_LEN
))
})?;
configs.push(Engine::port(
port.network_number,
mac,
port.transport.max_npdu(),
PortGroups {
groups: port.transport.group_destinations(),
local_broadcasts: port.transport.local_broadcast_destinations(),
},
));
limit_changes.push(port.transport.max_npdu_changes());
networks.push(port.network_number);
}
let engine = Engine::new(&configs, tokio_clock(), engine_options)
.map_err(|error| Error::Encoding(error.to_string()))?;
let network_control_ingress_sequence = Arc::new(AtomicU64::new(0));
let shared = Arc::new(Shared {
engine,
gate: control,
networks,
warnings: Default::default(),
});
let (send_txs, send_rxs): (Vec<_>, Vec<_>) = (0..ports.len())
.map(|_| send_queue::channel(send_queue::CAPACITY))
.unzip();
let replies = Arc::new(replies::ReplyLifetime::default());
let outputs = Arc::new(Outputs {
replies: Arc::clone(&replies),
send_txs,
local_tx,
local_control: local_control::LocalControl::new(
network_control,
Arc::clone(&network_control_ingress_sequence),
),
});
let sender_tasks = ports
.into_iter()
.zip(send_rxs)
.enumerate()
.map(|(index, (port, send_rx))| {
tokio::spawn(sender::run(
port.transport,
PortId(index as u8),
send_rx,
Arc::clone(&shared),
Arc::downgrade(&outputs),
))
})
.collect();
let limit_tasks = limit_changes
.into_iter()
.enumerate()
.filter_map(|(index, changes)| {
let port = PortId(index as u8);
let task = limits::follow(Arc::clone(&shared), port, changes?);
Some(tokio::spawn(task))
})
.collect();
shared.announce(&outputs.send_txs);
let dispatch_tasks = port_receivers
.into_iter()
.enumerate()
.map(|(index, rx)| {
let port = PortDispatch {
shared: Arc::clone(&shared),
outputs: Arc::clone(&outputs),
port: index as u8,
};
tokio::spawn(port.run(rx))
})
.collect();
let aging = Arc::clone(&shared);
let aging_task = tokio::spawn(async move {
let clock = aging.engine.clock();
let wake = |next: Option<Instant>| next.and_then(|at| clock.to_std(at));
let mut next = wake(aging.engine.lock().next_deadline());
loop {
match next {
Some(at) => tokio::time::sleep_until(at.into()).await,
None => std::future::pending().await,
}
next = wake(aging.sweep());
}
});
Ok((
Self {
replies,
outputs: Arc::downgrade(&outputs),
shared,
dispatch_tasks,
sender_tasks,
aging_task: Some(aging_task),
limit_tasks,
network_control_ingress_sequence,
},
local_rx,
counters,
))
}
pub fn table(&self) -> RouterTableGuard<'_> {
RouterTableGuard {
router: self.shared.engine.lock(),
}
}
pub fn now(&self) -> Instant {
self.shared.engine.now()
}
pub async fn stop(&mut self) {
self.replies.seal();
for task in self.dispatch_tasks.drain(..) {
task.abort();
let _ = task.await;
}
for task in self.sender_tasks.drain(..) {
task.abort();
let _ = task.await;
}
if let Some(task) = self.aging_task.take() {
task.abort();
let _ = task.await;
}
for task in self.limit_tasks.drain(..) {
task.abort();
let _ = task.await;
}
self.shared.engine.lock().cancel_discovery();
}
#[cfg(test)]
fn searches(&self) -> (usize, bool) {
let router = self.shared.engine.lock();
(router.discovery().len(), router.discovery().is_cancelled())
}
}
#[cfg(test)]
mod admission_tests;
#[cfg(test)]
mod bip_group_tests;
#[cfg(test)]
mod busy_scope_tests;
#[cfg(test)]
mod claim_tests;
#[cfg(test)]
mod congestion_tests;
#[cfg(test)]
mod corroboration_tests;
#[cfg(test)]
mod cutover_tests;
#[cfg(test)]
mod driven_port;
#[cfg(test)]
mod envelope_control_tests;
#[cfg(test)]
mod envelope_discovery_tests;
#[cfg(test)]
mod envelope_harness;
#[cfg(test)]
mod fixes_tests;
#[cfg(test)]
mod hardening_tests;
#[cfg(test)]
mod ingress_harness;
#[cfg(test)]
mod init_routing_table_tests;
#[cfg(test)]
mod late_refusal_tests;
#[cfg(test)]
mod limit_tests;
#[cfg(test)]
mod mstp_extended_tests;
#[cfg(test)]
mod originate_tests;
#[cfg(test)]
mod priority_harness;
#[cfg(test)]
mod priority_tests;
#[cfg(test)]
mod r7_reply_tests;
#[cfg(test)]
mod rb06_convergence_tests;
#[cfg(test)]
mod rb09_control_policy_tests;
#[cfg(test)]
mod reject_tests;
#[cfg(test)]
mod tests;
impl Drop for BACnetRouter {
fn drop(&mut self) {
self.replies.seal();
}
}