use std::collections::HashMap;
use std::net::Ipv4Addr;
#[cfg(feature = "ipv6")]
use std::net::Ipv6Addr;
use std::sync::Arc;
use std::time::Instant;
use bytes::{Bytes, BytesMut};
use tokio::sync::{broadcast, mpsc, Mutex};
use tokio::task::JoinHandle;
use tokio::time::{timeout, Duration};
use tracing::{debug, warn};
use bacnet_encoding::apdu::{
self, encode_apdu, validate_max_apdu_length, AbortPdu, Apdu,
ConfirmedRequest as ConfirmedRequestPdu, RejectPdu, SegmentAck as SegmentAckPdu, SimpleAck,
};
use bacnet_encoding::npdu::NpduAddress;
use bacnet_network::layer::NetworkLayer;
use bacnet_services::cov::COVNotificationRequest;
use bacnet_transport::bip::BipTransport;
#[cfg(feature = "ipv6")]
use bacnet_transport::bip6::Bip6Transport;
use bacnet_transport::port::TransportPort;
use bacnet_types::enums::{
ConfirmedServiceChoice, NetworkPriority, RejectReason, UnconfirmedServiceChoice,
};
use bacnet_types::error::Error;
use bacnet_types::MacAddr;
use crate::discovery::{DeviceTable, DeviceUpsertResult, DiscoveredDevice};
use crate::segmentation::{max_segment_payload, split_payload, SegmentReceiver, SegmentedPduType};
use crate::tsm::{Tsm, TsmConfig, TsmResponse};
pub const DEFAULT_COV_CHANNEL_CAPACITY: usize = 64;
pub const MAX_COV_CHANNEL_CAPACITY: usize = 65_536;
pub const DEVICE_EVENT_CHANNEL_CAPACITY: usize = 64;
const VALID_MAX_APDU_LENGTHS: [u16; 6] = [50, 128, 206, 480, 1024, 1476];
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum DeviceEventKind {
Discovered,
Updated,
Lost,
}
#[derive(Debug, Clone)]
pub struct DeviceEvent {
pub kind: DeviceEventKind,
pub device: DiscoveredDevice,
}
#[derive(Debug, Clone)]
pub struct ClientConfig {
pub interface: Ipv4Addr,
pub port: u16,
pub broadcast_address: Ipv4Addr,
pub apdu_timeout_ms: u64,
pub apdu_retries: u8,
pub max_apdu_length: u16,
pub max_segments: Option<u8>,
pub segmented_response_accepted: bool,
pub proposed_window_size: u8,
}
#[derive(Clone)]
pub struct ClientOptions {
pub cov_channel_capacity: usize,
confirmed_cov_notification_ack_policy: ConfirmedCOVNotificationAckPolicy,
}
impl Default for ClientConfig {
fn default() -> Self {
Self {
interface: Ipv4Addr::UNSPECIFIED,
port: 0xBAC0,
broadcast_address: Ipv4Addr::BROADCAST,
apdu_timeout_ms: 6000,
apdu_retries: 3,
max_apdu_length: 1476,
max_segments: None,
segmented_response_accepted: true,
proposed_window_size: 1,
}
}
}
impl Default for ClientOptions {
fn default() -> Self {
Self {
cov_channel_capacity: DEFAULT_COV_CHANNEL_CAPACITY,
confirmed_cov_notification_ack_policy:
cov_notifications::default_confirmed_cov_notification_ack_policy(),
}
}
}
impl ClientOptions {
pub fn with_cov_channel_capacity(mut self, capacity: usize) -> Self {
self.cov_channel_capacity = capacity;
self
}
fn validate(&self) -> Result<(), Error> {
if !(1..=MAX_COV_CHANNEL_CAPACITY).contains(&self.cov_channel_capacity) {
return Err(Error::Encoding(format!(
"invalid cov-channel-capacity {}; expected 1..={}",
self.cov_channel_capacity, MAX_COV_CHANNEL_CAPACITY
)));
}
Ok(())
}
}
pub struct ClientBuilder<T: TransportPort> {
config: ClientConfig,
options: ClientOptions,
transport: Option<T>,
}
impl<T: TransportPort + 'static> ClientBuilder<T> {
pub fn transport(mut self, transport: T) -> Self {
self.transport = Some(transport);
self
}
pub fn apdu_timeout_ms(mut self, ms: u64) -> Self {
self.config.apdu_timeout_ms = ms;
self
}
pub fn max_apdu_length(mut self, len: u16) -> Self {
self.config.max_apdu_length = len;
self
}
pub fn cov_channel_capacity(mut self, capacity: usize) -> Self {
self.options.cov_channel_capacity = capacity;
self
}
pub async fn build(self) -> Result<BACnetClient<T>, Error> {
let transport = self
.transport
.ok_or_else(|| Error::Encoding("transport not set on ClientBuilder".into()))?;
BACnetClient::start_with_options(self.config, transport, self.options).await
}
}
pub struct BipClientBuilder {
config: ClientConfig,
options: ClientOptions,
}
impl BipClientBuilder {
pub fn interface(mut self, ip: Ipv4Addr) -> Self {
self.config.interface = ip;
self
}
pub fn port(mut self, port: u16) -> Self {
self.config.port = port;
self
}
pub fn broadcast_address(mut self, addr: Ipv4Addr) -> Self {
self.config.broadcast_address = addr;
self
}
pub fn apdu_timeout_ms(mut self, ms: u64) -> Self {
self.config.apdu_timeout_ms = ms;
self
}
pub fn max_apdu_length(mut self, len: u16) -> Self {
self.config.max_apdu_length = len;
self
}
pub fn cov_channel_capacity(mut self, capacity: usize) -> Self {
self.options.cov_channel_capacity = capacity;
self
}
pub async fn build(self) -> Result<BACnetClient<BipTransport>, Error> {
let transport = BipTransport::new(
self.config.interface,
self.config.port,
self.config.broadcast_address,
);
BACnetClient::start_with_options(self.config, transport, self.options).await
}
}
const DEFAULT_BATCH_CONCURRENCY: usize = 32;
#[derive(Debug, Clone)]
pub struct DeviceReadRequest {
pub device_instance: u32,
pub object_identifier: bacnet_types::primitives::ObjectIdentifier,
pub property_identifier: bacnet_types::enums::PropertyIdentifier,
pub property_array_index: Option<u32>,
}
#[derive(Debug)]
pub struct DeviceReadResult {
pub device_instance: u32,
pub result: Result<bacnet_services::read_property::ReadPropertyACK, Error>,
}
#[derive(Debug, Clone)]
pub struct DeviceRpmRequest {
pub device_instance: u32,
pub specs: Vec<bacnet_services::rpm::ReadAccessSpecification>,
}
#[derive(Debug)]
pub struct DeviceRpmResult {
pub device_instance: u32,
pub result: Result<bacnet_services::rpm::ReadPropertyMultipleACK, Error>,
}
#[derive(Debug, Clone)]
pub struct DeviceWriteRequest {
pub device_instance: u32,
pub object_identifier: bacnet_types::primitives::ObjectIdentifier,
pub property_identifier: bacnet_types::enums::PropertyIdentifier,
pub property_array_index: Option<u32>,
pub property_value: Vec<u8>,
pub priority: Option<u8>,
}
#[derive(Debug)]
pub struct DeviceWriteResult {
pub device_instance: u32,
pub result: Result<(), Error>,
}
struct SegmentedReceiveState {
receiver: SegmentReceiver,
reply_mac: MacAddr,
expected_next_seq: u8,
last_activity: Instant,
window_position: u8,
proposed_window_size: u8,
}
const SEG_RECEIVER_TIMEOUT: Duration = Duration::from_secs(4);
type SegKey = (MacAddr, u8);
pub struct BACnetClient<T: TransportPort> {
config: ClientConfig,
network: Arc<NetworkLayer<T>>,
tsm: Arc<Mutex<Tsm>>,
device_table: Arc<Mutex<DeviceTable>>,
cov_tx: broadcast::Sender<ReceivedCOVNotification>,
device_tx: broadcast::Sender<DeviceEvent>,
dispatch_task: Option<JoinHandle<()>>,
seg_ack_senders: Arc<Mutex<HashMap<SegKey, mpsc::Sender<SegmentAckPdu>>>>,
local_mac: MacAddr,
}
impl BACnetClient<BipTransport> {
pub fn bip_builder() -> BipClientBuilder {
BipClientBuilder {
config: ClientConfig::default(),
options: ClientOptions::default(),
}
}
pub fn builder() -> BipClientBuilder {
Self::bip_builder()
}
pub async fn read_bdt(
&self,
target: &[u8],
) -> Result<Vec<bacnet_transport::bbmd::BdtEntry>, Error> {
self.network.transport().read_bdt(target).await
}
pub async fn write_bdt(
&self,
target: &[u8],
entries: &[bacnet_transport::bbmd::BdtEntry],
) -> Result<bacnet_types::enums::BvlcResultCode, Error> {
self.network.transport().write_bdt(target, entries).await
}
pub async fn read_fdt(
&self,
target: &[u8],
) -> Result<Vec<bacnet_transport::bbmd::FdtEntryWire>, Error> {
self.network.transport().read_fdt(target).await
}
pub async fn delete_fdt_entry(
&self,
target: &[u8],
ip: [u8; 4],
port: u16,
) -> Result<bacnet_types::enums::BvlcResultCode, Error> {
self.network
.transport()
.delete_fdt_entry(target, ip, port)
.await
}
pub async fn register_foreign_device_bvlc(
&self,
target: &[u8],
ttl: u16,
) -> Result<bacnet_types::enums::BvlcResultCode, Error> {
self.network
.transport()
.register_foreign_device_bvlc(target, ttl)
.await
}
}
#[cfg(feature = "ipv6")]
impl BACnetClient<Bip6Transport> {
pub fn bip6_builder() -> Bip6ClientBuilder {
Bip6ClientBuilder {
config: ClientConfig::default(),
options: ClientOptions::default(),
interface: Ipv6Addr::UNSPECIFIED,
device_instance: None,
}
}
}
#[cfg(feature = "ipv6")]
pub struct Bip6ClientBuilder {
config: ClientConfig,
options: ClientOptions,
interface: Ipv6Addr,
device_instance: Option<u32>,
}
#[cfg(feature = "ipv6")]
impl Bip6ClientBuilder {
pub fn interface(mut self, ip: Ipv6Addr) -> Self {
self.interface = ip;
self
}
pub fn port(mut self, port: u16) -> Self {
self.config.port = port;
self
}
pub fn device_instance(mut self, instance: u32) -> Self {
self.device_instance = Some(instance);
self
}
pub fn apdu_timeout_ms(mut self, ms: u64) -> Self {
self.config.apdu_timeout_ms = ms;
self
}
pub fn max_apdu_length(mut self, len: u16) -> Self {
self.config.max_apdu_length = len;
self
}
pub fn cov_channel_capacity(mut self, capacity: usize) -> Self {
self.options.cov_channel_capacity = capacity;
self
}
pub async fn build(self) -> Result<BACnetClient<Bip6Transport>, Error> {
let transport = Bip6Transport::new(self.interface, self.config.port, self.device_instance);
BACnetClient::start_with_options(self.config, transport, self.options).await
}
}
#[cfg(feature = "sc-tls")]
impl BACnetClient<bacnet_transport::sc::ScTransport<bacnet_transport::sc_tls::TlsWebSocket>> {
pub fn sc_builder() -> ScClientBuilder {
ScClientBuilder {
config: ClientConfig::default(),
options: ClientOptions::default(),
hub_url: String::new(),
tls_config: None,
vmac: [0; 6],
device_uuid: [0; 16],
heartbeat_interval_ms: 30_000,
heartbeat_timeout_ms: 60_000,
reconnect: None,
}
}
}
#[cfg(feature = "sc-tls")]
pub struct ScClientBuilder {
config: ClientConfig,
options: ClientOptions,
hub_url: String,
tls_config: Option<std::sync::Arc<tokio_rustls::rustls::ClientConfig>>,
vmac: bacnet_transport::sc_frame::Vmac,
device_uuid: [u8; 16],
heartbeat_interval_ms: u64,
heartbeat_timeout_ms: u64,
reconnect: Option<bacnet_transport::sc::ScReconnectConfig>,
}
#[cfg(feature = "sc-tls")]
impl ScClientBuilder {
pub fn hub_url(mut self, url: &str) -> Self {
self.hub_url = url.to_string();
self
}
pub fn tls_config(
mut self,
config: std::sync::Arc<tokio_rustls::rustls::ClientConfig>,
) -> Self {
self.tls_config = Some(config);
self
}
pub fn vmac(mut self, vmac: [u8; 6]) -> Self {
self.vmac = vmac;
self
}
pub fn device_uuid(mut self, uuid: [u8; 16]) -> Self {
self.device_uuid = uuid;
self
}
pub fn apdu_timeout_ms(mut self, ms: u64) -> Self {
self.config.apdu_timeout_ms = ms;
self
}
pub fn cov_channel_capacity(mut self, capacity: usize) -> Self {
self.options.cov_channel_capacity = capacity;
self
}
pub fn heartbeat_interval_ms(mut self, ms: u64) -> Self {
self.heartbeat_interval_ms = ms;
self
}
pub fn heartbeat_timeout_ms(mut self, ms: u64) -> Self {
self.heartbeat_timeout_ms = ms;
self
}
pub fn reconnect(mut self, config: bacnet_transport::sc::ScReconnectConfig) -> Self {
self.reconnect = Some(config);
self
}
fn validate_identity(&self) -> Result<(), Error> {
match self.vmac {
bacnet_transport::sc_frame::UNKNOWN_VMAC => Err(Error::Encoding(
"SC client builder: vmac must not be the reserved unknown VMAC".into(),
)),
bacnet_transport::sc_frame::BROADCAST_VMAC => Err(Error::Encoding(
"SC client builder: vmac must not be the reserved broadcast VMAC".into(),
)),
_ if self.device_uuid == [0; 16] => Err(Error::Encoding(
"SC client builder: device_uuid is required and must not be all zero".into(),
)),
_ => Ok(()),
}
}
fn sc_transport<W: bacnet_transport::sc::WebSocketPort>(
&self,
ws: W,
) -> bacnet_transport::sc::ScTransport<W> {
bacnet_transport::sc::ScTransport::new(ws, self.vmac)
.with_device_uuid(self.device_uuid)
.with_heartbeat_interval_ms(self.heartbeat_interval_ms)
.with_heartbeat_timeout_ms(self.heartbeat_timeout_ms)
}
#[cfg(test)]
pub(crate) async fn build_with_websocket_for_test<
W: bacnet_transport::sc::WebSocketPort + 'static,
>(
self,
ws: W,
) -> Result<BACnetClient<bacnet_transport::sc::ScTransport<W>>, Error> {
self.validate_identity()?;
self.options.validate()?;
let transport = self.sc_transport(ws);
BACnetClient::start_with_options(self.config, transport, self.options).await
}
pub async fn build(
self,
) -> Result<
BACnetClient<bacnet_transport::sc::ScTransport<bacnet_transport::sc_tls::TlsWebSocket>>,
Error,
> {
self.validate_identity()?;
self.options.validate()?;
let tls_config =
self.tls_config.as_ref().cloned().ok_or_else(|| {
Error::Encoding("SC client builder: tls_config is required".into())
})?;
let ws = bacnet_transport::sc_tls::TlsWebSocket::connect(&self.hub_url, tls_config.clone())
.await?;
let mut transport = self.sc_transport(ws);
if let Some(rc) = self.reconnect {
let hub_url = self.hub_url.clone();
let tls_config = tls_config.clone();
#[allow(deprecated)]
{
transport = transport
.with_connector(move || {
let hub_url = hub_url.clone();
let tls_config = tls_config.clone();
async move {
bacnet_transport::sc_tls::TlsWebSocket::connect(&hub_url, tls_config)
.await
}
})
.with_reconnect(rc);
}
}
BACnetClient::start_with_options(self.config, transport, self.options).await
}
}
#[derive(Clone, Copy)]
enum ConfirmedTarget<'a> {
Local {
mac: &'a [u8],
},
Routed {
router_mac: &'a [u8],
dest_network: u16,
dest_mac: &'a [u8],
},
}
impl<'a> ConfirmedTarget<'a> {
fn tsm_mac(&self) -> MacAddr {
match self {
Self::Local { mac } => MacAddr::from_slice(mac),
Self::Routed {
dest_network,
dest_mac,
..
} => routed_tsm_mac(*dest_network, dest_mac),
}
}
fn additional_npdu_header_len(&self) -> u16 {
match self {
Self::Local { .. } => 0,
Self::Routed { dest_mac, .. } => u16::try_from(dest_mac.len())
.unwrap_or(u16::MAX)
.saturating_add(4),
}
}
}
fn max_apdu_bucket_at_or_below(limit: u16) -> Option<u16> {
VALID_MAX_APDU_LENGTHS
.iter()
.rev()
.copied()
.find(|bucket| *bucket <= limit)
}
fn cap_max_apdu_to_transport(configured: u16, transport_limit: u16) -> Result<u16, Error> {
let limit = configured.min(transport_limit);
max_apdu_bucket_at_or_below(limit).ok_or_else(|| {
Error::Encoding(format!(
"transport max-APDU-length {transport_limit} leaves no valid BACnet APDU bucket"
))
})
}
fn routed_tsm_mac(network: u16, mac: &[u8]) -> MacAddr {
let mut key = MacAddr::new();
key.extend_from_slice(&[0xFF, b'R']);
key.extend_from_slice(&network.to_be_bytes());
key.push(mac.len() as u8);
key.extend_from_slice(mac);
key
}
fn response_tsm_mac(source_mac: &[u8], source_network: &Option<NpduAddress>) -> MacAddr {
match source_network {
Some(address) if !address.mac_address.is_empty() => {
routed_tsm_mac(address.network, &address.mac_address)
}
_ => MacAddr::from_slice(source_mac),
}
}
mod builder_options;
mod cov;
mod cov_notifications;
mod cov_renewal;
mod device_events;
mod device_mgmt;
mod discovery;
mod dispatch;
mod file_list;
mod lifecycle;
mod object_mgmt;
mod property;
mod requests;
mod segmentation;
pub use cov_notifications::{
COVNotificationDelivery, ConfirmedCOVNotificationAckPolicy, ConfirmedCOVNotificationResponse,
ReceivedCOVNotification,
};
pub use cov_renewal::{
ManagedCOVSubscription, ManagedCOVSubscriptionEvent, ManagedCOVSubscriptionOptions,
};
#[cfg(test)]
mod builder_options_tests;
#[cfg(test)]
mod cov_notification_tests;
#[cfg(test)]
mod cov_renewal_tests;
#[cfg(test)]
mod cov_tests;
#[cfg(test)]
mod device_events_tests;
#[cfg(all(test, feature = "sc-tls"))]
mod sc_builder_tests;
#[cfg(test)]
mod sc_max_apdu_tests;
#[cfg(test)]
mod segmentation_retransmit_tests;
#[cfg(test)]
mod tests;
impl<T: TransportPort + 'static> BACnetClient<T> {
pub fn generic_builder() -> ClientBuilder<T> {
ClientBuilder {
config: ClientConfig::default(),
options: ClientOptions::default(),
transport: None,
}
}
}