use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Weak};
use bacnet_encoding::apdu::{Apdu, decode_apdu};
use bacnet_endpoint_core::coordinator::{
AdmissionOutcome, CanonicalPeer, OutboundTransactionCoordinator,
};
pub use bacnet_endpoint_core::endpoint_ingress::EndpointApduDestination;
use bacnet_network::layer::ReceivedApdu;
use bacnet_transport::port::DataAttribute;
use bacnet_types::enums::PropertyIdentifier;
use bacnet_types::error::Error;
use bacnet_types::primitives::ObjectIdentifier;
use bacnet_client::EndpointRequester;
use bacnet_server::server::{
__endpoint_EndpointResponder as EndpointResponder,
__endpoint_NotificationTransactions as NotificationTransactions,
};
fn shutdown_error() -> Error {
Error::Encoding("endpoint shutdown".into())
}
#[doc(hidden)]
pub struct SessionToken {
open: AtomicBool,
pub coordinator: Arc<OutboundTransactionCoordinator>,
suspend_next: AtomicBool,
}
impl SessionToken {
#[doc(hidden)]
pub fn new(coordinator: Arc<OutboundTransactionCoordinator>) -> Arc<Self> {
Arc::new(Self {
open: AtomicBool::new(true),
coordinator,
suspend_next: AtomicBool::new(false),
})
}
#[doc(hidden)]
pub fn shutdown(&self) {
self.open.store(false, Ordering::Release);
}
#[doc(hidden)]
pub fn is_open(&self) -> bool {
self.open.load(Ordering::Acquire)
}
#[doc(hidden)]
pub fn suspend_next_reply(&self) {
self.suspend_next.store(true, Ordering::Release);
}
#[doc(hidden)]
pub fn take_suspend(&self) -> bool {
self.suspend_next.swap(false, Ordering::AcqRel)
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Commandability {
Commandable,
Noncommandable,
}
#[derive(Clone)]
pub struct ClientRoleHandle {
token: Weak<SessionToken>,
requester: EndpointRequester,
bip_broadcast: Option<std::net::Ipv4Addr>,
source_audit: Option<Weak<crate::source_audit::SourceAudit>>,
}
impl ClientRoleHandle {
#[doc(hidden)]
pub fn new(token: &Arc<SessionToken>, requester: EndpointRequester) -> Self {
Self {
token: Arc::downgrade(token),
requester,
bip_broadcast: None,
source_audit: None,
}
}
pub(crate) fn with_bip_broadcast(mut self, broadcast: Option<std::net::Ipv4Addr>) -> Self {
self.bip_broadcast = broadcast;
self
}
#[doc(hidden)]
pub fn validate_write_property(
&self,
request: &bacnet_services::write_property::WritePropertyRequest,
) -> Result<(), Error> {
self.requester
.validate_operation(&bacnet_client::EndpointOperationRequest::Write(
request.clone(),
))
}
pub async fn write_property(
&self,
destination_mac: &[u8],
request: bacnet_services::write_property::WritePropertyRequest,
commandability: Commandability,
) -> Result<(), Error> {
self.check_open()?;
let broadcast = self
.bip_broadcast
.ok_or_else(|| Error::Encoding("endpoint WriteProperty requires IPv4 B/IP".into()))?;
let (ip, port) = bacnet_transport::bvll::decode_bip_mac(destination_mac)?;
let ip = std::net::Ipv4Addr::from(ip);
if ip.is_unspecified()
|| ip.is_multicast()
|| ip.is_broadcast()
|| ip == broadcast
|| port == 0
{
return Err(Error::Encoding(
"endpoint WriteProperty requires direct B/IP IPv4 unicast".into(),
));
}
let destination = EndpointApduDestination::Direct {
destination_mac: bacnet_types::MacAddr::from_slice(destination_mac),
};
if let Some(source) = &self.source_audit {
source
.upgrade()
.ok_or_else(shutdown_error)?
.write(&self.requester, destination, request, commandability)
.await
} else {
self.requester
.prepare_write(destination, Vec::new(), request)
.await?
.execute()
.await
.result?
.into_write()
}
}
pub(crate) fn with_source_audit(
mut self,
source: &Arc<crate::source_audit::SourceAudit>,
) -> Self {
self.source_audit = Some(Arc::downgrade(source));
self
}
fn session(&self) -> Result<Arc<SessionToken>, Error> {
self.token.upgrade().ok_or_else(shutdown_error)
}
fn check_open(&self) -> Result<Arc<SessionToken>, Error> {
let session = self.session()?;
if !session.is_open() {
return Err(shutdown_error());
}
Ok(session)
}
pub async fn read_property(
&self,
destination_mac: &[u8],
object_identifier: ObjectIdentifier,
property_identifier: PropertyIdentifier,
property_array_index: Option<u32>,
) -> Result<bacnet_services::read_property::ReadPropertyACK, Error> {
self.read_property_with_destination(
EndpointApduDestination::Direct {
destination_mac: bacnet_types::MacAddr::from_slice(destination_mac),
},
Vec::new(),
object_identifier,
property_identifier,
property_array_index,
)
.await
}
pub async fn read_property_with_destination(
&self,
destination: EndpointApduDestination,
data_attributes: Vec<DataAttribute>,
object_identifier: ObjectIdentifier,
property_identifier: PropertyIdentifier,
property_array_index: Option<u32>,
) -> Result<bacnet_services::read_property::ReadPropertyACK, Error> {
self.check_open()?;
if let Some(source) = &self.source_audit {
let source = source.upgrade().ok_or_else(shutdown_error)?;
return source
.read(
&self.requester,
destination,
data_attributes,
bacnet_client::EndpointReadRequest::Property(
bacnet_services::read_property::ReadPropertyRequest {
object_identifier,
property_identifier,
property_array_index,
},
),
)
.await?
.into_property();
}
self.requester
.read_property_with_destination(
destination,
data_attributes,
object_identifier,
property_identifier,
property_array_index,
)
.await
}
pub async fn read_property_multiple(
&self,
destination_mac: &[u8],
specs: Vec<bacnet_types::constructed::ReadAccessSpecification>,
) -> Result<bacnet_services::rpm::ReadPropertyMultipleACK, Error> {
self.read_property_multiple_with_destination(
EndpointApduDestination::Direct {
destination_mac: bacnet_types::MacAddr::from_slice(destination_mac),
},
Vec::new(),
specs,
)
.await
}
pub async fn read_property_multiple_with_destination(
&self,
destination: EndpointApduDestination,
data_attributes: Vec<DataAttribute>,
specs: Vec<bacnet_types::constructed::ReadAccessSpecification>,
) -> Result<bacnet_services::rpm::ReadPropertyMultipleACK, Error> {
self.check_open()?;
let request = bacnet_client::EndpointReadRequest::Multiple(
bacnet_services::rpm::ReadPropertyMultipleRequest {
list_of_read_access_specs: specs,
},
);
if let Some(source) = &self.source_audit {
source
.upgrade()
.ok_or_else(shutdown_error)?
.read(&self.requester, destination, data_attributes, request)
.await?
.into_multiple()
} else {
self.requester
.prepare_read(destination, data_attributes, request)
.await?
.execute()
.await
.result?
.into_multiple()
}
}
pub async fn read_range(
&self,
destination_mac: &[u8],
object_identifier: ObjectIdentifier,
property_identifier: PropertyIdentifier,
property_array_index: Option<u32>,
range: Option<bacnet_services::read_range::RangeSpec>,
) -> Result<bacnet_services::read_range::ReadRangeAck, Error> {
self.read_range_with_destination(
EndpointApduDestination::Direct {
destination_mac: bacnet_types::MacAddr::from_slice(destination_mac),
},
Vec::new(),
object_identifier,
property_identifier,
property_array_index,
range,
)
.await
}
pub async fn read_range_with_destination(
&self,
destination: EndpointApduDestination,
data_attributes: Vec<DataAttribute>,
object_identifier: ObjectIdentifier,
property_identifier: PropertyIdentifier,
property_array_index: Option<u32>,
range: Option<bacnet_services::read_range::RangeSpec>,
) -> Result<bacnet_services::read_range::ReadRangeAck, Error> {
let request = bacnet_services::read_range::ReadRangeRequest {
object_identifier,
property_identifier,
property_array_index,
range,
};
self.read_range_checked(
destination,
data_attributes,
request,
bacnet_services::read_range::ReadRangeValidation::Strict,
)
.await
.map(|reply| reply.ack)
}
pub async fn read_range_with(
&self,
destination_mac: &[u8],
request: &bacnet_services::read_range::ReadRangeRequest,
validation: bacnet_services::read_range::ReadRangeValidation,
) -> Result<bacnet_services::read_range::ReadRangeReply, Error> {
self.read_range_checked(
EndpointApduDestination::Direct {
destination_mac: bacnet_types::MacAddr::from_slice(destination_mac),
},
Vec::new(),
request.clone(),
validation,
)
.await
}
pub async fn read_log_page(
&self,
destination_mac: &[u8],
log: ObjectIdentifier,
cursor: bacnet_client::log_reader::LogCursor,
page_size: u16,
) -> Result<bacnet_client::log_reader::LogPage, Error> {
bacnet_client::log_reader::read_log_page(self, destination_mac, log, cursor, page_size)
.await
}
async fn read_range_checked(
&self,
destination: EndpointApduDestination,
data_attributes: Vec<DataAttribute>,
request: bacnet_services::read_range::ReadRangeRequest,
validation: bacnet_services::read_range::ReadRangeValidation,
) -> Result<bacnet_services::read_range::ReadRangeReply, Error> {
self.check_open()?;
if let Some(source) = &self.source_audit {
let source = source.upgrade().ok_or_else(shutdown_error)?;
return source
.read(
&self.requester,
destination,
data_attributes,
bacnet_client::EndpointReadRequest::Range(request, validation),
)
.await?
.into_range();
}
self.requester
.read_range_with_destination(destination, data_attributes, request, validation)
.await
}
#[doc(hidden)]
pub async fn complete_pre_admitted(
&self,
admission: bacnet_endpoint_core::coordinator::Admission,
apdu: Apdu,
received: ReceivedApdu,
) -> bool {
if self.check_open().is_err() {
return false;
}
self.requester
.complete_pre_admitted(admission, apdu, received)
.await
}
#[allow(dead_code)]
pub(crate) fn close_for_owner(&self) {
self.requester.close();
}
}
impl bacnet_client::log_reader::LogRequester for ClientRoleHandle {
async fn read_unsigned(
&self,
mac: &[u8],
log: ObjectIdentifier,
property: PropertyIdentifier,
) -> Result<u64, Error> {
let ack = self.read_property(mac, log, property, None).await?;
bacnet_client::log_reader::unsigned_value(&ack.property_value)
}
async fn read_range_lenient(
&self,
mac: &[u8],
request: &bacnet_services::read_range::ReadRangeRequest,
) -> Result<bacnet_services::read_range::ReadRangeReply, Error> {
self.read_range_with(
mac,
request,
bacnet_services::read_range::ReadRangeValidation::Lenient,
)
.await
}
}
#[derive(Clone)]
pub struct ServerRoleHandle {
token: Weak<SessionToken>,
responder: Arc<EndpointResponder>,
notifications: Arc<NotificationTransactions>,
}
impl ServerRoleHandle {
#[doc(hidden)]
pub fn new(
token: &Arc<SessionToken>,
responder: Arc<EndpointResponder>,
notifications: Arc<NotificationTransactions>,
) -> Self {
Self {
token: Arc::downgrade(token),
responder,
notifications,
}
}
fn check_open(&self) -> Result<(), Error> {
let session = self.token.upgrade().ok_or_else(shutdown_error)?;
if !session.is_open() {
return Err(shutdown_error());
}
Ok(())
}
pub fn is_session_alive(&self) -> bool {
self.token.upgrade().is_some_and(|s| s.is_open())
}
pub fn suspend_next_reply(&self) -> Result<(), Error> {
let session = self.token.upgrade().ok_or_else(shutdown_error)?;
if !session.is_open() {
return Err(shutdown_error());
}
session.suspend_next_reply();
Ok(())
}
pub async fn handle_inbound(&self, mut received: ReceivedApdu) -> Result<bool, Error> {
self.check_open()?;
if !received.provenance.is_direct_peer()
&& received.direct_response.is_none()
&& received.reply_tx.is_some()
&& self.token.upgrade().is_some_and(|s| s.take_suspend())
{
let _ = received.reply_tx.take();
}
self.responder.handle(received).await
}
pub fn complete_notification_pre_admitted(
&self,
admission: bacnet_endpoint_core::coordinator::Admission,
apdu: &Apdu,
) -> bool {
if self.check_open().is_err() {
return false;
}
self.notifications.complete_pre_admitted(admission, apdu)
}
#[allow(dead_code)]
pub(crate) fn close_for_owner(&self) {
self.responder.close();
self.notifications.close();
}
}
fn assert_send_sync<T: Send + Sync>() {}
const _: fn() = || {
assert_send_sync::<ClientRoleHandle>();
assert_send_sync::<ServerRoleHandle>();
};
#[doc(hidden)]
pub fn inbound_canonical_peer(
received: &ReceivedApdu,
local_network: Option<u16>,
) -> CanonicalPeer {
CanonicalPeer::from_source(
received.source_mac.as_slice(),
received.source_network.as_ref(),
local_network,
)
}
#[doc(hidden)]
pub fn decode_terminal(received: &ReceivedApdu) -> Option<Apdu> {
decode_apdu(received.apdu.clone()).ok()
}
#[doc(hidden)]
pub fn is_requester_lease(admission: &bacnet_endpoint_core::coordinator::Admission) -> bool {
admission.metadata().owner() == bacnet_endpoint_core::coordinator::LeaseOwner::Requester
}
#[doc(hidden)]
pub fn admit_once(
coordinator: &OutboundTransactionCoordinator,
received: &ReceivedApdu,
local_network: Option<u16>,
apdu: &Apdu,
) -> Result<AdmissionOutcome, bacnet_endpoint_core::coordinator::CoordinatorError> {
coordinator.admit_from_source(
received.source_mac.as_slice(),
received.source_network.as_ref(),
local_network,
apdu,
)
}