mod actions;
mod application;
mod application_feed;
mod application_replies;
mod application_transactions;
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;
mod shutdown;
mod startup;
pub use application::{RouterApplicationPort, RouterApplicationPorts};
pub use replies::RouterResponse;
pub use shutdown::{RouterStopError, RouterStopFailure};
use std::ops::{Deref, DerefMut};
use std::sync::Arc;
use std::sync::atomic::AtomicU64;
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::MacAddr;
use bacnet_types::error::Error;
use bytes::Bytes;
pub use counters::Counters;
pub use originate::OriginateFailure;
use tokio::sync::mpsc;
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::{Engine, EngineGuard, PORTS, tokio_clock};
pub use options::{LocalApduReceiver, NetworkControlReceiver, RouterOptions, StartedRouter};
use port_groups::PortGroups;
#[cfg(test)]
use {
bacnet_encoding::npdu::{Npdu, encode_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 = bacnet_transport::any::AnyTransport> {
pub transport: T,
pub network_number: u16,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct RouterPortInfo {
index: u8,
network_number: u16,
mac: MacAddr,
}
impl RouterPortInfo {
pub fn index(&self) -> u8 {
self.index
}
pub fn network_number(&self) -> u16 {
self.network_number
}
pub fn local_mac(&self) -> &[u8] {
&self.mac
}
}
pub struct BACnetRouter {
replies: Arc<replies::ReplyLifetime>,
outputs: std::sync::Weak<Outputs>,
shared: Arc<Shared>,
lifecycle: shutdown::Lifecycle,
port_info: Vec<RouterPortInfo>,
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
}
pub fn ports(&self) -> &[RouterPortInfo] {
&self.port_info
}
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) -> Result<(), RouterStopError> {
self.replies.seal();
self.shared.engine.lock().cancel_discovery();
self.lifecycle.stop().await
}
#[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 owned_bip_tests;
#[cfg(test)]
mod owned_fixture;
#[cfg(test)]
mod owned_lifecycle_tests;
#[cfg(test)]
mod owned_startup_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();
self.shared.engine.lock().cancel_discovery();
self.lifecycle.begin();
}
}