Current section
Files
Jump to
Current section
Files
lib/bacnet/stack/client.ex
defmodule BACnet.Stack.Client do
@moduledoc """
The BACnet client is responsible for connecting the application to the BACnet transport protocol
and vice versa - it interfaces with the BACnet transport protocol, using the transport behaviour.
The client will take requests and send them to the BACnet transport protocol and ultimately listen for
frames from the BACnet transport protocol.
The application will receive process messages on the specified `notification_receiver`. If there's no
notification receiver and a confirmed service request is received, the client will automatically send
a Reject APDU with reason `:other` to the remote BACnet device,
the application won't be informed in any way.
BACnet BVLL and BACnet NPDU are directly forwarded to the application, without any processing.
For BACnet BVLL, the following message is sent:
```elixir
{:bacnet_transport, protocol_id, source_address, {:bvlc, bvlc}, portal}
```
This is the same as `t:BACnet.Stack.TransportBehaviour.transport_msg/0`.
For BACnet NSPDU, the following message is sent:
```elixir
{:bacnet_transport, protocol_id, source_address, {:network, bvlc, npci, nsdu}, portal}
```
This is the same as `t:BACnet.Stack.TransportBehaviour.transport_msg/0`.
For BACnet APDU, the following message is sent:
```elixir
{:bacnet_client, reference() | nil, apdu, {source_address, bvlc, npci}, pid()}
```
The reference is only present on confirmed service requests and is used for `reply/4`.
APDU is `t:BACnet.Protocol.apdu/0`. The PID is of the `Client` process.
For the rest, see `t:BACnet.Stack.TransportBehaviour.transport_msg/0`.
BACnet APDUs are decoded, checked against the internal cache and forwarded to the application.
The client has an internal cache for requests for the application and replies from the application
to the transport protocol in order to deduplicate request (i.e. when the application takes for the
remote BACnet device far too long and re-sends the APDU).
In case of duplicated request, the request will not be forwarded to the application,
instead it will be silently dropped and the application will possibly reply in a timely fashion,
or the APDU timeout will occur and the Client will reply instead - this situation may arise when
the response does not arrive at the remote BACnet device within
the remote BACnet device's configured APDU timeout window.
Individual APDU timeouts on a per-source basis can be applied using `set_apdu_timeouts/2`.
The client will also keep track of sent APDUs for confirmed service requests and re-send them
automatically, if no response arrives in the APDU timeout timeframe. If the maximum APDU retries
is reached, the request will be deleted and the application will get an APDU timeout response.
Invoke IDs are automatically managed on a per destination (with device ID, if given) basis to avoid
duplicated invoke IDs being sent to the same destination.
This mechanism can be disabled on startup and allows external management (or usage with care) if desired.
If the application takes too long to respond to a remote BACnet request, the client will automatically
send an Abort APDU and respond to a `reply` request from the application with an app timeout response.
If the application replies to a routed request (Forwarded NPDU), the client will automatically
set the correct destination address in `reply/4`.
The BACnet client will not automatically respond to Who-Is, Who-Has, Time-Synchronization, etc. queries,
as this is outside of the responsibility of this low level BACnet client implementation.
"""
alias BACnet.Protocol
alias BACnet.Protocol.APDU
alias BACnet.Protocol.APDU.ConfirmedServiceRequest
alias BACnet.Protocol.BvlcForwardedNPDU
alias BACnet.Protocol.Constants
alias BACnet.Protocol.IncompleteAPDU
alias BACnet.Protocol.NPCI
alias BACnet.Protocol.NpciTarget
alias BACnet.Stack.EncoderProtocol
alias BACnet.Stack.Segmentator
alias BACnet.Stack.SegmentsStore
alias BACnet.Stack.Telemetry
alias BACnet.Stack.TransportBehaviour
import BACnet.Internal, only: [is_server: 1, log_debug: 1]
require Constants
require Logger
use GenServer
@apdu_timer_offset 50
@apdu_timeout_multiplicator 0.9
@call_timeout Application.compile_env(:bacstack, :client_call_timeout, 60_000)
@retry_send_after_time Application.compile_env(:bacstack, :client_retry_send_after_time, 1000)
@default_apdu_retries 3
@default_apdu_timeout 3000
@default_window_size 16
defguardp is_apdu(type)
when type in [
APDU.Abort,
APDU.ComplexACK,
APDU.ConfirmedServiceRequest,
APDU.Error,
APDU.Reject,
APDU.SegmentACK,
APDU.SimpleACK,
APDU.UnconfirmedServiceRequest
]
defguardp is_apdu_resp(type)
when type in [
APDU.Abort,
APDU.ComplexACK,
APDU.Error,
APDU.Reject,
APDU.SegmentACK,
APDU.SimpleACK
]
defmodule ApduTimer do
@moduledoc """
Internal module for `BACnet.Stack.Client`.
It is used as APDU timer for outgoing APDUs.
It holds together all the necessary information
to track the APDU, time, retry count and contains
information that is used to reply to the application.
"""
@typedoc """
Representative type for its purpose.
"""
@type t :: %__MODULE__{
portal: TransportBehaviour.portal(),
destination: term(),
device_id: non_neg_integer() | nil,
apdu: Protocol.apdu(),
send_opts: Keyword.t(),
call_ref: term(),
retry_count: non_neg_integer(),
timer: reference(),
monotonic_time: integer()
}
@fields [
:portal,
:destination,
:device_id,
:apdu,
:send_opts,
:call_ref,
:retry_count,
:monotonic_time,
:timer
]
@enforce_keys @fields
defstruct @fields
end
defmodule ReplyTimer do
@moduledoc """
Internal module for `BACnet.Stack.Client`.
It is used as reply timer for incoming APDUs.
It holds together all the necessary information
to fire when the application does not respond
fast enough and will reply negatively to the
remote BACnet client.
"""
@typedoc """
Representative type for its purpose.
"""
@type t :: %__MODULE__{
bvlc: Protocol.bvlc(),
npci: NPCI.t(),
portal: TransportBehaviour.portal(),
service_req: Protocol.apdu(),
source_addr: term(),
device_id: non_neg_integer() | nil,
ref: reference(),
monotonic_time: integer(),
has_retried: boolean(),
timer: reference() | nil
}
@fields [
:bvlc,
:npci,
:portal,
:service_req,
:source_addr,
:device_id,
:ref,
:monotonic_time,
:has_retried,
:timer
]
@enforce_keys @fields
defstruct @fields
end
defmodule State do
@moduledoc """
Internal module for `BACnet.Stack.Client`.
It is used as `GenServer` state.
"""
@typedoc """
Key for the application reply timer.
"""
@type app_timer_key ::
{address :: term(), device_id :: non_neg_integer() | nil, invoke_id :: byte()}
@typedoc """
Representative type for its purpose.
"""
@type t :: %__MODULE__{
apdu_timers: %{optional(app_timer_key()) => BACnet.Stack.Client.ApduTimer.t()},
apdu_timeouts: BACnet.Stack.Client.apdu_timeouts(),
app_reply_mapping: %{
optional(reference()) => app_timer_key()
},
app_reply_timers: %{
optional(app_timer_key()) => BACnet.Stack.Client.ReplyTimer.t()
},
notification_receiver: [Process.dest()],
segmentator: Segmentator.server(),
segments_store: SegmentsStore.server(),
transport_broadcast_addr: term(),
transport_mod: module(),
transport_pid: TransportBehaviour.transport(),
transport_portal: TransportBehaviour.portal(),
opts: %{
apdu_retries: non_neg_integer(),
apdu_timeout: pos_integer(),
disable_app_timeout: boolean(),
disable_invoke_id_management: boolean(),
npci_source: NpciTarget.t() | nil,
segmented_rcv_window_overwrite: boolean(),
supervisor_mod: module()
}
}
@fields [
:apdu_timers,
:apdu_timeouts,
:app_reply_mapping,
:app_reply_timers,
:notification_receiver,
:segmentator,
:segments_store,
:transport_broadcast_addr,
:transport_mod,
:transport_pid,
:transport_portal,
:opts
]
@enforce_keys @fields
defstruct @fields
end
@typedoc """
Per-source APDU timeouts.
Device ID is only known, if the source transmits it with the APDU,
as such, most of the time it can be nil.
"""
@type apdu_timeouts :: %{
optional({source_address :: term(), device_id :: non_neg_integer() | nil}) =>
apdu_timeout :: non_neg_integer()
}
@typedoc """
Valid start options. For a description of each, see `start_link/1`.
"""
@type start_option ::
{:apdu_retries, pos_integer()}
| {:apdu_timeout, pos_integer()}
| {:disable_app_timeout, boolean()}
| {:disable_invoke_id_management, boolean()}
| {:notification_receiver, Process.dest() | [Process.dest()]}
| {:npci_source, NpciTarget.t()}
| {:segmentator, Segmentator.server()}
| {:segments_store, SegmentsStore.server()}
| {:segmented_rcv_window_overwrite, boolean()}
| {:transport, module() | {module(), TransportBehaviour.transport()}}
| GenServer.option()
@typedoc """
Represents a server process of the Client module.
"""
@type server :: GenServer.server()
@typedoc """
List of start options.
"""
@type start_options :: [start_option()]
@doc """
Starts and links the BACnet Client.
The following options are available, in addition to `t:GenServer.options/0`:
- `apdu_retries: pos_integer()` - Optional. The amount of APDU sending retries (defaults to 3).
Only applied to confirmed service requests.
- `apdu_timeout: pos_integer()` - Optional. The APDU timeout to be waiting for a response, in ms (defaults to 3000ms).
Only applied to confirmed service requests.
- `disable_app_timeout: boolean()` - Optional. Disables the application timeout mechanism.
- `disable_invoke_id_management: boolean()` - Optional. Disables `invoke_id` management and override in request payloads.
- `notification_receiver: Process.dest() | [Process.dest()]` - Optional. The destination to send messages to.
- `npci_source: NpciTarget.t()` - Optional. The NPCI target to use as source for outgoing APDUs.
- `segmentator: Segmentator.server()` - Required. The segmentator server to use.
- `segments_store: SegmentsStore.server()` - Required. The segments store server to use.
- `segmented_rcv_window_overwrite: boolean()` - Optional. Enable to overwrite the window size to 1 for incoming
segmented APDUs when it is bound to be routed (i.e. subject to BACnet/IP UDP packet re-ordering).
If you're having difficulty receiving segmented APDUs and the packets get routed on BACnet/IP,
you should consider enabling this and see if it helps.
- `transport: module() | {module(), TransportBehaviour.transport()}` - Required. The transport to use.
`module` is equivalent to `{module, module}` (the module name is registered process name).
The given transport must implement the `BACnet.Stack.TransportBehaviour` behaviour.
"""
@spec start_link(start_options()) :: GenServer.on_start()
def start_link(opts) do
unless Keyword.keyword?(opts) do
raise ArgumentError, "start_link/1 expected a keyword list, got: #{inspect(opts)}"
end
{opts2, genserver_opts} =
Keyword.split(opts, [
:apdu_retries,
:apdu_timeout,
:disable_app_timeout,
:disable_invoke_id_management,
:notification_receiver,
:npci_source,
:segmentator,
:segments_store,
:segmented_rcv_window_overwrite,
:transport
])
validate_start_link_opts(opts2)
GenServer.start_link(__MODULE__, Map.new(opts2), genserver_opts)
end
@doc """
Add a source to the per-source APDU timeouts map. This is only used for receiving.
Each source is identified by source address and device ID (device ID
is only known if the source transmit it in the BACnet NPCI).
"""
@spec add_apdu_timeout(server(), term(), non_neg_integer() | nil, non_neg_integer()) :: :ok
def add_apdu_timeout(server, source_address, device_id, timeout)
when is_server(server) and
((is_integer(device_id) and device_id in 1..4_194_302) or is_nil(device_id)) and
is_integer(timeout) and timeout > 0 do
GenServer.call(server, {:add_apdu_timeout, source_address, device_id, timeout})
end
@doc """
Get the per-source APDU timeouts map. This is only used for receiving.
Each source is identified by source address and device ID (device ID
is only known if the source transmit it in the BACnet NPCI).
"""
@spec get_apdu_timeouts(server()) :: {:ok, apdu_timeouts()}
def get_apdu_timeouts(server) when is_server(server) do
GenServer.call(server, :get_apdu_timeouts)
end
@doc """
Get the transport used in the client.
"""
@spec get_transport(server()) ::
{module(), TransportBehaviour.transport(), TransportBehaviour.portal()}
def get_transport(server) when is_server(server) do
GenServer.call(server, :get_transport)
end
@doc """
Replies to a confirmed service request from a remote BACnet device.
The APDU frame may be segmentated by the client, depending on the
APDU size and maximum transmittable APDU size.
The reference identifies the request in the client. The request is hold
in the client to be able to apply application timeout constraints and
automatically respond on application timeout.
The remote BACnet device may send the same request again within the
configured APDU timeout and thus will be silently deduplicated (dropped).
The reference is given as part of the BACnet client notification process message.
If an automatic application timeout response has been sent (Abort APDU),
`{:error, :app_timeout}` will be returned when trying to reply to the
request.
See `send/4` for more information about `opts`.
The options `:max_apdu_length`, `:max_segments` and `:segmentation_supported` of `opts`
are automatically derived from the confirmed service request, if not explicitely given.
"""
@spec reply(server(), reference(), Protocol.apdu(), Keyword.t()) ::
:ok
| {:error, :app_timeout}
| {:error, term()}
| {:error, {Exception.t(), stacktrace :: Exception.stacktrace()}}
def reply(server, ref, %type{} = reply, opts \\ [])
when is_server(server) and is_reference(ref) and is_apdu_resp(type) and is_list(opts) do
unless Keyword.keyword?(opts) do
raise ArgumentError,
"reply/4 expected opts to be a keyword list, " <>
"got: #{inspect(opts)}"
end
validate_send_opts(opts, "reply/4")
GenServer.call(server, {:reply, ref, reply, opts}, @call_timeout)
end
@doc """
Remove a source from the per-source APDU timeouts map. This is only used for receiving.
Each source is identified by source address and device ID (device ID
is only known if the source transmit it in the BACnet NPCI).
"""
@spec remove_apdu_timeout(server(), term(), non_neg_integer() | nil) :: :ok
def remove_apdu_timeout(server, source_address, device_id)
when is_server(server) and
((is_integer(device_id) and device_id in 1..4_194_302) or is_nil(device_id)) do
GenServer.call(server, {:remove_apdu_timeout, source_address, device_id})
end
@doc """
Sends the given APDU frame to the specified destination (remote BACnet device).
The APDU frame may be segmentated by the client.
The client will keep track of sent confirmed service requests and automatically
re-send the APDUs, if the APDU times out, without a response from the remote
BACnet server. If the maximum APDU retry count gets reached,
`{:error, :apdu_timeout}` will be returned.
This function returns, for confirmed service requests, after receiving the
response from the remote BACnet server, for everything else almost immediately.
As such, this function will block for maximum 60s (default compiled value), before
the backpressure mechanism will exit the caller.
Destination depends on the transport module and is validated against the
transport module.
BACnet Abort/Error/Reject are also returned in `:ok` tuples, not only
acknowledgements and requests.
When sending and the APDU is too large and thus is needed to be segmented,
the client will check accordinging to the given options,
whether segmentation can occur and how many segments are supported.
If the remote device does not support segmentation or a buffer overflow
would occur due to too many segments, this client will send an Abort APDU
to the remote device and return an error to the caller. The same will happen
for too long APDUs that can not be segmented.
See the `c:BACnet.Stack.TransportBehaviour.send/4` documentation for what `opts` can be.
In addition the following are available:
- `device_id: pos_integer()` - Optional. The remote BACnet device ID.
Only used for invoke ID management together with the destination address.
Specifying it allows invoke IDs to be used on a per device ID basis,
if multiple devices run on the same destination address (i.e. MS/TP to IP gateway).
Please note, if wrongfully used, this may lead to collisions and invalid data -
including replies sent to requests that were never meant for that request.
- `max_apdu_length: pos_integer()` - Optional. The maximum APDU length
the remote BACnet device supports (defaults to the transport max APDU length).
- `max_segments: pos_integer()` - Optional. Maximum amount of segments the
remote BACnet device can accept (defaults to 2).
- `segmentation_supported: Constants.segmentation()` - Optional. Which kind
of segmentation the remote BACnet device supports (defaults to none).
"""
@spec send(server(), term(), Protocol.apdu(), Keyword.t()) ::
:ok
| {:ok, Protocol.apdu()}
| {:error, :apdu_timeout}
| {:error, :apdu_too_long}
| {:error, :segmentation_not_supported}
| {:error, term()}
| {:error, {Exception.t(), stacktrace :: Exception.stacktrace()}}
| {:error, {term(), stacktrace :: Exception.stacktrace()}}
def send(server, destination, %type{} = data, opts \\ [])
when is_server(server) and is_apdu(type) and is_list(opts) do
unless Keyword.keyword?(opts) do
raise ArgumentError,
"send/4 expected opts to be a keyword list, " <>
"got: #{inspect(opts)}"
end
validate_send_opts(opts, "send/4")
GenServer.call(server, {:send, destination, data, opts}, @call_timeout)
end
@doc """
Set the per-source APDU timeouts map. This is only used for receiving.
Each source is identified by source address and device ID (device ID
is only known if the source transmit it in the BACnet NPCI).
"""
@spec set_apdu_timeouts(server(), apdu_timeouts()) :: :ok
def set_apdu_timeouts(server, %{} = timeouts) when is_server(server) do
GenServer.call(server, {:set_apdu_timeouts, timeouts})
end
@doc """
Puts the subscriber in the `notification_receiver` list.
The list contains only unique elements, so this function call is idempotent.
After this function returns, the subscriber will start to receive
process messages as lined out by the module documentation.
If `subscriber` is a PID, it will be monitored and automatically removed.
This means for short lived processes, using the PID is recommended
as the PID is automatically removed when the process dies.
"""
@spec subscribe(server(), pid() | Process.dest() | GenServer.server()) :: :ok
def subscribe(server, subscriber) when is_server(server) and is_server(subscriber) do
GenServer.call(server, {:subscribe, subscriber})
end
@doc """
Removes the subscriber from the `notification_receiver` list.
After this function returns, the subscriber will stop receiving
process messages as lined out by the module documentation.
"""
@spec unsubscribe(server(), pid() | Process.dest() | GenServer.server()) :: :ok
def unsubscribe(server, subscriber) when is_server(server) and is_server(subscriber) do
GenServer.call(server, {:unsubscribe, subscriber})
end
@doc false
def init(opts) do
# Transform transport option into transport module and pid/name
{transport_mod, transport_pid} =
case opts.transport do
{mod, pid} -> {mod, pid}
mod -> {mod, mod}
end
state_opts =
opts
|> Map.drop([
:notification_receiver,
:segmentator,
:segments_store,
:transport
])
|> Map.put_new(:apdu_retries, @default_apdu_retries)
|> Map.put_new(:apdu_timeout, @default_apdu_timeout)
|> Map.put_new(:disable_app_timeout, false)
|> Map.put_new(:disable_invoke_id_management, false)
|> Map.put_new(:npci_source, nil)
|> Map.put_new(:segmented_rcv_window_overwrite, false)
state = %State{
apdu_timers: %{},
apdu_timeouts: %{},
app_reply_mapping: %{},
app_reply_timers: %{},
notification_receiver: List.wrap(Map.get(opts, :notification_receiver, [])),
segmentator: opts.segmentator,
segments_store: opts.segments_store,
transport_broadcast_addr: transport_mod.get_broadcast_address(transport_pid),
transport_mod: transport_mod,
transport_pid: transport_pid,
transport_portal: transport_mod.get_portal(transport_pid),
opts: state_opts
}
log_debug(fn -> "Client: Started on #{inspect(self())}" end)
{:ok, state}
end
@doc false
def handle_call(
{:add_apdu_timeout, source_address, device_id, timeout},
_from,
%State{apdu_timeouts: apdu_timeouts} = state
) do
log_debug(fn -> "Client: Received add_apdu_timeout request" end)
new_state = %{
state
| apdu_timeouts: Map.put(apdu_timeouts, {source_address, device_id}, timeout)
}
{:reply, :ok, new_state}
end
def handle_call(:get_apdu_timeouts, _from, %State{} = state) do
log_debug(fn -> "Client: Received get_apdu_timeouts request" end)
{:reply, {:ok, state.apdu_timeouts}, state}
end
def handle_call(
{:remove_apdu_timeout, source_address, device_id},
_from,
%State{apdu_timeouts: apdu_timeouts} = state
) do
log_debug(fn -> "Client: Received remove_apdu_timeout request" end)
new_state = %{state | apdu_timeouts: Map.delete(apdu_timeouts, {source_address, device_id})}
{:reply, :ok, new_state}
end
def handle_call({:set_apdu_timeouts, %{} = new_value}, _from, %State{} = state) do
log_debug(fn -> "Client: Received set_apdu_timeouts request" end)
new_state = %{state | apdu_timeouts: new_value}
{:reply, :ok, new_state}
end
def handle_call(:get_transport, _from, %State{} = state) do
log_debug(fn -> "Client: Received get_transport request" end)
{:reply, {state.transport_mod, state.transport_pid, state.transport_portal}, state}
end
def handle_call(
{:transport_call, function, args},
_from,
%State{transport_mod: trans_mod} = state
)
when is_atom(function) and is_list(args) do
log_debug(fn -> "Client: Received transport_call request" end)
result =
try do
case apply(trans_mod, function, args) do
{:ok, _term} = term -> term
{:error, _err} = term -> term
term -> {:ok, term}
end
catch
_kind, e -> {:error, {e, __STACKTRACE__}}
end
{:reply, result, state}
end
def handle_call(:get_broadcast_address, _from, %State{transport_mod: trans_mod} = state) do
log_debug(fn -> "Client: Received get_broadcast_address request" end)
result =
try do
brd = trans_mod.get_broadcast_address(state.transport_pid)
{:ok, brd}
catch
_kind, e -> {:error, {e, __STACKTRACE__}}
end
{:reply, result, state}
end
def handle_call({:subscribe, pid}, _from, %State{} = state) do
log_debug(fn -> "Client: Received subscribe request" end)
new_state =
Map.update!(state, :notification_receiver, fn list ->
if Enum.member?(list, pid) do
list
else
# Monitor PIDs, because they should be removed
# once the subscribing process dies
if is_pid(pid) do
Process.monitor(pid)
end
[pid | list]
end
end)
{:reply, :ok, new_state}
end
def handle_call({:unsubscribe, pid}, _from, %State{} = state) do
log_debug(fn -> "Client: Received unsubscribe request" end)
new_state =
Map.update!(state, :notification_receiver, fn list ->
List.delete(list, pid)
end)
{:reply, :ok, new_state}
end
def handle_call(
{:send, destination, %{} = data, opts},
from,
%State{transport_mod: trans_mod} = state
) do
log_debug(fn -> "Client: Received send request" end)
try do
trans_mod.is_valid_destination(destination)
catch
_kind, e -> {:reply, {:error, {e, __STACKTRACE__}}, state}
else
is_valid_dest ->
if is_valid_dest do
if state.opts.disable_invoke_id_management and
Map.has_key?(data, :invoke_id) and
Map.has_key?(
state.apdu_timers,
{destination, opts[:device_id], Map.get(data, :invoke_id)}
) do
{:reply, {:error, :duplicate_invoke_id}, state}
else
needs_tracking = is_struct(data, ConfirmedServiceRequest)
case send_data(
data,
opts,
destination,
state.transport_portal,
opts[:device_id],
needs_tracking,
from,
state,
false
) do
{:ok, new_state} ->
if needs_tracking, do: {:noreply, new_state}, else: {:reply, :ok, new_state}
{:retry, new_state} ->
# We don't have a free invoke_id, so retry later
Process.send_after(
self(),
{:"$gen_call", from, {:send, destination, data, opts}},
@retry_send_after_time
)
{:noreply, new_state}
term ->
{:reply, term, state}
end
end
else
{:reply, {:error, :invalid_destination}, state}
end
end
end
def handle_call({:reply, ref, %{} = data, opts}, _from, %State{} = state) do
log_debug(fn -> "Client: Received reply request" end)
reply_key = Map.get(state.app_reply_mapping, ref, ref)
{reply, new_state} =
case Map.fetch(state.app_reply_timers, reply_key) do
{:ok, reply} ->
destination =
case reply.bvlc do
%BvlcForwardedNPDU{} = forwarded ->
{forwarded.originating_ip, forwarded.originating_port}
_else ->
reply.source_addr
end
new_opts =
opts
|> Keyword.put_new(:max_apdu_length, reply.service_req.max_apdu)
|> Keyword.put_new_lazy(:max_segments, fn ->
case reply.service_req.max_segments do
# BTL-2017: Unspecified might be as low as 2 and is kept for backwards compatibility
:unspecified -> 2
val -> val
end
end)
|> Keyword.put_new_lazy(:segmentation_supported, fn ->
if reply.service_req.segmented_response_accepted do
Constants.macro_assert_name(:segmentation, :segmented_receive)
end
end)
Telemetry.execute_client_inc_apdu_reply(self(), data, new_opts, reply, state)
case send_data(
data,
new_opts,
destination,
reply.portal,
reply.device_id,
false,
nil,
state,
true
) do
{:ok, new_state} ->
if reply.timer, do: Process.cancel_timer(reply.timer)
new_state2 =
new_state
|> Map.update!(:app_reply_mapping, fn mapping ->
Map.delete(mapping, reply_key)
end)
|> Map.update!(:app_reply_timers, fn timers ->
Map.delete(timers, reply_key)
end)
{:ok, new_state2}
{:error, _err} = err ->
{err, state}
end
:error ->
{{:error, :app_timeout}, state}
end
{:reply, reply, new_state}
end
def handle_call(_msg, _from, state) do
{:noreply, state}
end
@doc false
def handle_cast(_msg, state) do
{:noreply, state}
end
@doc false
def handle_info(
{:bacnet_transport, _proto, source_address, {:bvlc, _bvlc}, _portal} = data,
%State{notification_receiver: dest} = state
) do
log_debug(fn ->
"Client: Got BACnet stack BVLC data from #{inspect(source_address)}, data: #{inspect(data)}"
end)
Telemetry.execute_client_transport_message(self(), data, state)
send_process_dest(dest, data)
{:noreply, state}
end
def handle_info(
{:bacnet_transport, _proto, source_address, {:network, _bvlc, _npci, _nsdu}, _portal} =
data,
%State{notification_receiver: dest} = state
) do
log_debug(fn ->
"Client: Got BACnet stack NSDU data from #{inspect(source_address)}, data: #{inspect(data)}"
end)
Telemetry.execute_client_transport_message(self(), data, state)
send_process_dest(dest, data)
{:noreply, state}
end
def handle_info(
{:bacnet_transport, _proto, source_address, {:apdu, bvlc, npci, raw_apdu} = cb_data,
portal} = data,
%State{} = state
) do
log_debug(fn ->
"Client: Got BACnet stack APDU data from #{inspect(source_address)}, data: #{inspect(cb_data)}"
end)
Telemetry.execute_client_transport_message(self(), data, state)
new_state = handle_raw_apdu(source_address, bvlc, npci, raw_apdu, portal, state)
{:noreply, new_state}
end
def handle_info({:apdu_timer, key}, %State{opts: %{apdu_retries: max_retry}} = state) do
# If remote server takes too long to reply, send APDU again
# for n times and if still no reply, abort and reply to call
log_debug(fn -> "Client: Received APDU timer message for #{inspect(key)}" end)
new_state =
case Map.fetch(state.apdu_timers, key) do
{:ok, %ApduTimer{retry_count: ^max_retry} = timer} ->
log_debug(fn ->
"Client: APDU timer #{inspect(key)} has reached max retry count, removing"
end)
Telemetry.execute_client_request_apdu_timer(self(), timer, state)
GenServer.reply(timer.call_ref, {:error, :apdu_timeout})
%{state | apdu_timers: Map.delete(state.apdu_timers, key)}
{:ok, %ApduTimer{} = timer} ->
case send_data(
timer.apdu,
timer.send_opts,
timer.destination,
timer.portal,
timer.device_id,
# No tracking because we update the timer "inline"
false,
nil,
state,
# Skip check because we "lock" the invoke ID using the APDU timer
true
) do
{:ok, new_state} ->
%{
new_state
| apdu_timers:
Map.update!(state.apdu_timers, key, fn timer ->
%{
timer
| retry_count: timer.retry_count + 1,
timer:
Process.send_after(
self(),
{:apdu_timer, key},
state.opts.apdu_timeout + @apdu_timer_offset
)
}
end)
}
{:error, err} ->
log_debug(fn ->
"Client: Error on trying to re-send APDU on APDU timeout, error: #{inspect(err)}"
end)
state
end
:error ->
state
end
{:noreply, new_state}
end
def handle_info({:reply_timer, reply_key}, %State{} = state) do
# If app takes too long to reply to request, send abort APDU to remote device
log_debug(fn -> "Client: Received reply timer message for #{inspect(reply_key)}" end)
new_state =
case Map.fetch(state.app_reply_timers, reply_key) do
{:ok, %ReplyTimer{} = reply} ->
apdu = %APDU.Abort{
sent_by_server: true,
invoke_id: reply.service_req.invoke_id,
reason: Constants.macro_assert_name(:abort_reason, :application_exceeded_reply_time)
}
Telemetry.execute_client_inc_apdu_timeout(self(), reply, state)
case send_data(
apdu,
get_reply_opts_for_npci(apdu, :original_unicast, reply.npci, state),
reply.source_addr,
reply.portal,
reply.device_id,
false,
nil,
state,
true
) do
{:ok, new_state} ->
%{
new_state
| app_reply_mapping: Map.delete(state.app_reply_mapping, reply.ref),
app_reply_timers: Map.delete(state.app_reply_timers, reply_key)
}
{:error, err} ->
log_debug(fn ->
"Client: Encountered error during app reply timeout when sending APDU abort reply, error: #{inspect(err)}"
end)
if reply.has_retried do
# We have already retried sending the Abort APDU, we do not do it again,
# clean up regardless
%{
state
| app_reply_mapping: Map.delete(state.app_reply_mapping, reply.ref),
app_reply_timers: Map.delete(state.app_reply_timers, reply_key)
}
else
# Retry in 10ms again once
timers =
Map.put(state.app_reply_timers, reply_key, %{
reply
| timer: Process.send_after(self(), {:reply_timer, reply_key}, 10),
has_retried: true
})
%{state | app_reply_timers: timers}
end
end
:error ->
state
end
{:noreply, new_state}
end
def handle_info({:DOWN, _ref, :process, pid, _reason}, %State{} = state) do
# Remove dead processes from our notification receiver list
new_state =
Map.update!(state, :notification_receiver, fn list ->
List.delete(list, pid)
end)
{:noreply, new_state}
end
# def handle_info({:DOWN, _ref, _type, _object, _reason}, %State{} = state)
# when object == state.transport_pid do
# # When transport goes down, go down too - we need to get initialized from start
# {:stop, :transport_down, state}
# end
def handle_info(_msg, state) do
{:noreply, state}
end
@spec handle_raw_apdu(
term(),
Protocol.bvlc(),
NPCI.t(),
binary(),
TransportBehaviour.portal(),
State.t()
) :: State.t()
defp handle_raw_apdu(
source_address,
bvlc,
%NPCI{} = npci,
raw_apdu,
portal,
%State{transport_mod: trans_mod} = state
) do
case APDU.decode(raw_apdu) do
{:ok, apdu} ->
Telemetry.execute_client_inc_apdu(self(), source_address, bvlc, npci, apdu, state)
handle_apdu(source_address, bvlc, npci, apdu, portal, state)
{:incomplete, %IncompleteAPDU{} = incomplete} ->
log_debug(fn ->
"Client: Received fragmented APDU from #{inspect(source_address)}, " <>
"invoke_id: #{incomplete.invoke_id}, seq_number: #{incomplete.sequence_number}, " <>
"more_follows: #{inspect(incomplete.more_follows)}"
end)
incomplete_apdu =
if state.opts.segmented_rcv_window_overwrite and
trans_mod.is_destination_routed(state.transport_pid, source_address) do
IncompleteAPDU.set_window_size(incomplete, 1)
else
incomplete
end
case SegmentsStore.segment(
state.segments_store,
incomplete_apdu,
trans_mod,
portal,
source_address,
destination: npci.source
) do
{:ok, complete_data} ->
log_debug(fn ->
"Client: Completed fragmented APDU from #{inspect(source_address)}, invoke_id: #{incomplete.invoke_id}"
end)
Telemetry.execute_client_inc_apdu_segmentation_completed(
self(),
source_address,
bvlc,
npci,
raw_apdu,
complete_data,
incomplete_apdu,
state
)
handle_raw_apdu(source_address, bvlc, npci, complete_data, portal, state)
:incomplete ->
Telemetry.execute_client_inc_apdu_segmentation_incomplete(
self(),
source_address,
bvlc,
npci,
raw_apdu,
incomplete_apdu,
state
)
state
{:error, err, cancelled} ->
log_debug(fn ->
"Client: Got error from segments store for #{inspect(source_address)}, invoke_id: #{incomplete.invoke_id}, " <>
"cancelled: #{inspect(cancelled)}, error: #{inspect(err)}"
end)
Telemetry.execute_client_inc_apdu_segmentation_error(
self(),
source_address,
bvlc,
npci,
raw_apdu,
incomplete_apdu,
err,
cancelled,
state
)
state
end
{:error, err} ->
Telemetry.execute_client_inc_apdu_decode_error(
self(),
source_address,
bvlc,
npci,
raw_apdu,
err,
state
)
if is_tuple(err) and is_atom(elem(err, 0)) and
String.starts_with?("#{elem(err, 0)}", "unknown_") do
log_debug(fn ->
"Client: Encountered unknown enumeration (constant) while decoding APDU, " <>
if(npci.expects_reply, do: "sending Reject APDU to source, ", else: "") <>
"error: #{inspect(err)}"
end)
if npci.expects_reply do
send_reply_for_unknown_enumeration(
source_address,
bvlc,
npci,
raw_apdu,
portal,
state
)
else
state
end
else
log_debug(fn -> "Client: Got invalid APDU data, error: #{inspect(err)}" end)
state
end
end
# rescue
# e ->
# log_debug(fn ->
# "Client: Encountered exception while decoding APDU, " <>
# "exception: #{inspect(Exception.message(e))}"
# end)
# state
end
@spec handle_apdu(
term(),
Protocol.bvlc(),
NPCI.t(),
Protocol.apdu(),
TransportBehaviour.portal(),
State.t()
) :: State.t()
defp handle_apdu(source_address, bvlc, npci, apdu, portal, state)
# Do not handle APDU and send Reject APDU if no listener configured and reply expected
# Unless this is a reply to one of the requests sent
defp handle_apdu(
source_address,
bvlc,
%NPCI{expects_reply: true} = npci,
%apdu_type{invoke_id: invoke_id} = apdu,
portal,
%State{apdu_timers: apdu_timers, notification_receiver: []} = state
)
when not is_apdu_resp(apdu_type) or map_size(apdu_timers) == 0 do
log_debug(fn ->
"Client: Received APDU with reply expected and no listener, " <>
"sending Reject APDU to source"
end)
# Get the BACnet device ID, if available from NPCI source
device_id =
case npci.source do
%NpciTarget{address: adr} when adr != nil -> adr
_else -> nil
end
reject = %APDU.Reject{
invoke_id: invoke_id,
reason: Constants.macro_assert_name(:reject_reason, :other)
}
Telemetry.execute_client_inc_apdu_rejected(
self(),
source_address,
bvlc,
npci,
reject,
apdu,
state
)
case send_data(
reject,
get_reply_opts_for_npci(reject, bvlc, npci, state),
source_address,
portal,
device_id,
false,
nil,
state,
true
) do
{:ok, new_state} ->
new_state
{:error, err} ->
log_debug(fn ->
"Client: Encountered error on sending Reject reply due to no listener, error: #{inspect(err)}"
end)
state
end
end
# Do not handle APDU if there's no listener configured
# Usually no reply is expected from us, but might be "expected" due to response
# Unless this is a reply to one of the requests sent
defp handle_apdu(
_source_address,
_bvlc,
%NPCI{} = _npci,
%apdu_type{} = _apdu,
_portal,
%State{apdu_timers: apdu_timers, notification_receiver: []} = state
)
when not is_apdu_resp(apdu_type) or map_size(apdu_timers) == 0 do
state
end
# Handle APDU since we have a listener
defp handle_apdu(source_address, bvlc, %NPCI{} = npci, apdu, portal, %State{} = state) do
# Get the BACnet device ID, if available from NPCI source
device_id =
case npci.source do
%NpciTarget{address: adr} when adr != nil -> adr
_else -> nil
end
# Get the invoke_id from the APDU (only relevant for confirmed requests)
invoke_id = Map.get(apdu, :invoke_id)
reply_key = {source_address, device_id, invoke_id}
# If confirmed service request, look up and see if it already exists, if it does, ignore it
# If it does not exist yet, create an app reply timer to send Abort APDU to client on timeout
{handle, ref, new_state} =
check_for_duplicated_requests(
device_id,
source_address,
bvlc,
npci,
apdu,
reply_key,
portal,
state
)
# case state.notification_receiver do
# [] -> :ok
# dest -> send_process_dest(dest, {:bacnet_client_info, self(), :apdu, bvlc, npci, {apdu, portal, ref}})
# end
log_debug(fn ->
dev_id = if device_id, do: ":DEV-#{inspect(device_id)}"
keyword = if handle, do: "handled", else: "ignored"
"Client: Received APDU #{inspect(apdu.__struct__)} from #{inspect(source_address)}#{dev_id} to be #{keyword}"
end)
if handle do
Telemetry.execute_client_inc_apdu_handled(self(), source_address, bvlc, npci, apdu, state)
do_handle_apdu(ref, source_address, bvlc, npci, apdu, portal, device_id, new_state)
else
Telemetry.execute_client_inc_apdu_duplicated(
self(),
source_address,
bvlc,
npci,
apdu,
state
)
new_state
end
end
# If confirmed service request, look up and see if it already exists, if it does, ignore it
# If it does not exist yet, create an app reply timer to send abort APDU to client on timeout
@spec check_for_duplicated_requests(
non_neg_integer() | nil,
term(),
Protocol.bvlc(),
NPCI.t(),
Protocol.apdu(),
term(),
term(),
State.t()
) :: {should_handle :: boolean(), reply_id :: reference() | nil, new_state :: State.t()}
defp check_for_duplicated_requests(
device_id,
source_address,
bvlc,
npci,
apdu,
reply_key,
portal,
state
)
defp check_for_duplicated_requests(
device_id,
source_address,
bvlc,
%NPCI{} = npci,
%ConfirmedServiceRequest{} = apdu,
reply_key,
portal,
%State{} = state
) do
case Map.fetch(state.app_reply_timers, reply_key) do
# Duplicated request (same request by source_address, device_id, invoke_id)
{:ok, _reply} ->
log_debug(fn ->
dev_id = if device_id, do: ":DEV-#{inspect(device_id)}"
"Client: Received duplicated request from " <>
inspect(source_address) <> "#{dev_id} with invoke ID #{apdu.invoke_id}"
end)
{false, nil, state}
:error ->
# Get timeout or use default APDU timeout, and multiply with factor - the rest is our reserve to reply
timeout =
trunc(
Enum.find_value(state.apdu_timeouts, state.opts.apdu_timeout, fn
{{^source_address, ^device_id}, value} -> value
{{^source_address, nil}, value} -> value
_else -> nil
end) * @apdu_timeout_multiplicator
)
put_app_reply_timer(
reply_key,
%ReplyTimer{
bvlc: bvlc,
npci: npci,
portal: portal,
service_req: apdu,
source_addr: source_address,
device_id: device_id,
ref: make_ref(),
has_retried: false,
monotonic_time: System.monotonic_time(),
timer:
unless(state.opts.disable_app_timeout,
do: Process.send_after(self(), {:reply_timer, reply_key}, timeout)
)
},
state
)
end
end
defp check_for_duplicated_requests(
_device_id,
_source_address,
_bvlc,
_npci,
_apdu,
_reply_key,
_portal,
%State{} = state
),
do: {true, nil, state}
@spec put_app_reply_timer(
term(),
ReplyTimer.t(),
State.t()
) :: {boolean(), reference(), State.t()}
defp put_app_reply_timer(reply_key, reply_timer, %State{} = state) do
{true, reply_timer.ref,
%{
state
| app_reply_mapping: Map.put(state.app_reply_mapping, reply_timer.ref, reply_key),
app_reply_timers: Map.put(state.app_reply_timers, reply_key, reply_timer)
}}
end
@spec do_handle_apdu(
reference() | nil,
term(),
Protocol.bvlc(),
NPCI.t(),
Protocol.apdu(),
TransportBehaviour.portal(),
non_neg_integer() | nil,
State.t()
) :: State.t()
defp do_handle_apdu(ref, source_address, bvlc, npci, apdu, portal, device_id, state)
defp do_handle_apdu(
ref,
source_address,
bvlc,
%NPCI{} = npci,
%ConfirmedServiceRequest{} = apdu,
portal,
device_id,
%State{} = state
) do
send_notification_apdu(
ref,
source_address,
bvlc,
npci,
apdu,
portal,
device_id,
state
)
end
defp do_handle_apdu(
ref,
source_address,
bvlc,
%NPCI{} = npci,
%APDU.UnconfirmedServiceRequest{} = apdu,
portal,
device_id,
%State{} = state
) do
send_notification_apdu(
ref,
source_address,
bvlc,
npci,
apdu,
portal,
device_id,
state
)
end
defp do_handle_apdu(
ref,
source_address,
bvlc,
%NPCI{} = npci,
%type{} = apdu,
portal,
device_id,
%State{} = state
)
when type in [APDU.ComplexACK, APDU.SimpleACK] do
case remove_apdu_timer_for_response(source_address, apdu, device_id, state) do
{nil, new_state} ->
send_notification_apdu(
ref,
source_address,
bvlc,
npci,
apdu,
portal,
device_id,
new_state
)
{%ApduTimer{} = timer, new_state} ->
Telemetry.execute_client_request_stop(
self(),
source_address,
bvlc,
npci,
apdu,
timer,
state
)
GenServer.reply(timer.call_ref, {:ok, apdu})
new_state
end
end
defp do_handle_apdu(
_ref,
source_address,
_bvlc,
%NPCI{} = _npci,
%APDU.SegmentACK{} = apdu,
_portal,
_device_id,
%State{} = state
) do
# Send SegmentACK to Segmentator for processing
case Segmentator.handle_apdu(state.segmentator, source_address, apdu) do
:ok ->
:ok
{:error, err} ->
log_debug(fn ->
"Client: Error while handling SegmentACK by Segmentator, error: #{inspect(err)}"
end)
Telemetry.execute_client_exception(
self(),
:error,
err,
[Telemetry.make_stacktrace_from_env(__ENV__)],
%{},
state
)
end
state
end
defp do_handle_apdu(
ref,
source_address,
bvlc,
%NPCI{} = npci,
%type{} = apdu,
portal,
device_id,
%State{} = state
)
when type in [APDU.Abort, APDU.Error, APDU.Reject] do
# Send the APDU to the Segmentator and SegmentsStore process
Segmentator.handle_apdu(state.segmentator, source_address, apdu)
SegmentsStore.cancel(state.segments_store, source_address, apdu)
case remove_apdu_timer_for_response(source_address, apdu, device_id, state) do
{nil, new_state} ->
send_notification_apdu(
ref,
source_address,
bvlc,
npci,
apdu,
portal,
device_id,
new_state
)
{%ApduTimer{} = timer, new_state} ->
Telemetry.execute_client_request_stop(
self(),
source_address,
bvlc,
npci,
apdu,
timer,
state
)
GenServer.reply(timer.call_ref, {:ok, apdu})
new_state
end
end
@spec remove_apdu_timer_for_response(
term(),
Protocol.apdu(),
non_neg_integer() | nil,
State.t()
) ::
{ApduTimer.t() | nil, State.t()}
defp remove_apdu_timer_for_response(
source_address,
%{invoke_id: invoke_id} = _apdu,
device_id,
state
) do
reply_key = {source_address, device_id, invoke_id}
case Map.fetch(state.apdu_timers, reply_key) do
{:ok, %ApduTimer{} = timer} ->
Process.cancel_timer(timer.timer)
{timer, %{state | apdu_timers: Map.delete(state.apdu_timers, reply_key)}}
_else ->
{nil, state}
end
end
@spec send_notification_apdu(
reference() | nil,
term(),
Protocol.bvlc(),
NPCI.t(),
Protocol.apdu(),
TransportBehaviour.portal(),
non_neg_integer() | nil,
State.t()
) :: State.t()
defp send_notification_apdu(
ref,
source_address,
bvlc,
%NPCI{} = npci,
apdu,
_portal,
_device_id,
%State{notification_receiver: dest} = state
) do
send_process_dest(dest, {:bacnet_client, ref, apdu, {source_address, bvlc, npci}, self()})
state
end
@spec send_process_dest(Process.dest() | [Process.dest()], term()) :: :ok
defp send_process_dest(dest, msg) when is_list(dest) do
Enum.each(dest, &Kernel.send(&1, msg))
catch
# Ignore any exception coming from send/2 (an "invalid" destination raises! [i.e. an atom but it's not registered])
_type, _err -> :ok
end
# We do this here at compile time, so we have at runtime no need to recompute the same thing (hot path)
# It is not a MapSet because MapSet.to_list/1 gets always called, so using Map iterator is cheaper
@new_invoke_id_mapset Map.new(Enum.to_list(0..255//1), fn key -> {key, nil} end)
@spec find_free_invoke_id(term(), non_neg_integer() | nil, State.t()) ::
{:ok, byte()} | :error
defp find_free_invoke_id(destination, device_id, state)
# Optimize case where apdu timers is empty
defp find_free_invoke_id(_destination, _device_id, %State{apdu_timers: timers} = _state)
when map_size(timers) == 0,
do: {:ok, 0}
defp find_free_invoke_id(destination, device_id, %State{} = state) do
free_ids =
Enum.reduce(state.apdu_timers, @new_invoke_id_mapset, fn
{_key,
%ApduTimer{destination: ^destination, device_id: ^device_id, apdu: %{invoke_id: id}}} =
_timer,
acc ->
Map.delete(acc, id)
_else, acc ->
acc
end)
if map_size(free_ids) > 0 do
free_ids
|> :maps.iterator()
|> :maps.next()
|> elem(0)
|> then(&{:ok, &1})
else
:error
end
end
@spec send_data(
Protocol.apdu(),
Keyword.t(),
term(),
TransportBehaviour.portal(),
non_neg_integer() | nil,
boolean(),
term() | nil,
State.t(),
boolean()
) :: {:ok, new_state :: State.t()} | {:error, term()} | {:retry, new_state :: State.t()}
defp send_data(
apdu,
opts,
destination,
portal,
device_id,
needs_tracking,
call_ref,
state,
skip_invoke_id_check
)
defp send_data(
%{invoke_id: _id} = apdu,
opts,
destination,
portal,
device_id,
needs_tracking,
call_ref,
%State{opts: %{disable_invoke_id_management: false}} = state,
false
) do
case find_free_invoke_id(destination, device_id, state) do
{:ok, new_invoke_id} ->
send_data(
%{apdu | invoke_id: new_invoke_id},
opts,
destination,
portal,
device_id,
needs_tracking,
call_ref,
state,
true
)
:error ->
{:retry, state}
end
end
defp send_data(
apdu,
opts,
destination,
portal,
device_id,
needs_tracking,
call_ref,
%State{transport_mod: trans_mod} = state,
_skip_invoke_id_check
) do
sys_mono_time = System.monotonic_time()
try do
# Catch any errors when trying to encode the APDU
bin = EncoderProtocol.encode(apdu)
# 50 is the minimum APDU size each device needs to support
max_apdu_len = min(max(opts[:max_apdu_length] || 0, 50), trans_mod.max_npdu_length())
{bin, max_apdu_len}
catch
kind, e ->
Telemetry.execute_client_exception(
self(),
kind,
e,
__STACKTRACE__,
%{apdu: apdu, destination: destination},
state
)
{:error, {e, __STACKTRACE__}}
else
{apdu_data, max_apdu_len0} ->
send_opts =
opts
|> Keyword.drop([:max_apdu_length, :max_segments, :segmentation_supported])
|> kw_put_new(:source, state.opts.npci_source)
|> Keyword.put_new(:is_broadcast, state.transport_broadcast_addr == destination)
# Do basic NPCI size calculation and subtract it from the max APDU size
# 6 = APCI header, 2 = NPCI header
max_apdu_len =
max_apdu_len0 - 6 - 2 - if(send_opts[:source], do: 9, else: 0) -
if(send_opts[:destination], do: 10, else: 0)
apdu_length = IO.iodata_length(apdu_data)
max_segments = opts[:max_segments] || 2
apdu_too_long = apdu_length > max_apdu_len
apdu_supports_seg = EncoderProtocol.supports_segmentation(apdu)
supports_segments = supports_segmentation(opts[:segmentation_supported])
result =
cond do
apdu_too_long and not apdu_supports_seg ->
# If segmentation is not supported by the APDU type,
# send an abort and indiciate Abort APDU_TOO_LONG,
# but only if this is not a request to a remote device
resp =
if EncoderProtocol.is_response(apdu) do
abort = %APDU.Abort{
sent_by_server: true,
invoke_id: apdu.invoke_id,
reason: Constants.macro_assert_name(:abort_reason, :apdu_too_long)
}
Telemetry.execute_client_send_error(
self(),
destination,
apdu,
send_opts,
abort,
:apdu_too_long,
state
)
trans_mod.send(portal, destination, abort, send_opts)
else
:ok
end
# Return the send error if present, or our error (no error swallowing)
with :ok <- resp do
{:error, :apdu_too_long}
end
apdu_too_long and not supports_segments ->
# Segmentation not supported by the remote device,
# indicate Abort SEGMENTATION_NOT_SUPPORTED,
# but only if this is a response,
# if this is a request, we do not need to send anything
resp =
if EncoderProtocol.is_response(apdu) do
abort = %APDU.Abort{
sent_by_server: true,
invoke_id: apdu.invoke_id,
reason:
Constants.macro_assert_name(:abort_reason, :segmentation_not_supported)
}
Telemetry.execute_client_send_error(
self(),
destination,
apdu,
send_opts,
abort,
:apdu_segmentation_unsupported,
state
)
trans_mod.send(portal, destination, abort, send_opts)
else
:ok
end
# Return the send error if present, or our error (no error swallowing)
with :ok <- resp do
{:error, :segmentation_not_supported}
end
apdu_too_long ->
Telemetry.execute_client_send(
self(),
destination,
apdu,
send_opts,
true,
state
)
Segmentator.create_sequence(
state.segmentator,
{trans_mod, state.transport_pid, portal},
destination,
%{
apdu
| proposed_window_size:
Map.get(apdu, :proposed_window_size) || @default_window_size
},
max_apdu_len,
max_segments,
send_opts
)
true ->
Telemetry.execute_client_send(
self(),
destination,
apdu,
send_opts,
false,
state
)
trans_mod.send(portal, destination, apdu_data, send_opts)
end
if needs_tracking do
case result do
:ok ->
key = {destination, device_id, Map.get(apdu, :invoke_id)}
# Do basic APDU segments count estimation
factor = if(apdu_too_long, do: apdu_length / (max_apdu_len - 5) + 1, else: 1)
timer = %ApduTimer{
portal: portal,
destination: destination,
device_id: device_id,
apdu: apdu,
send_opts: opts,
call_ref: call_ref,
retry_count: 0,
monotonic_time: sys_mono_time,
timer:
Process.send_after(
self(),
{:apdu_timer, key},
state.opts.apdu_timeout * factor + @apdu_timer_offset
)
}
Telemetry.execute_client_request_start(
self(),
destination,
apdu,
send_opts,
timer,
state
)
new_state = %{state | apdu_timers: Map.put(state.apdu_timers, key, timer)}
{:ok, new_state}
{:error, _err} = err ->
err
end
else
case result do
:ok -> {:ok, state}
term -> term
end
end
end
end
@spec send_reply_for_unknown_enumeration(
term(),
Protocol.bvlc(),
NPCI.t(),
binary(),
TransportBehaviour.portal(),
State.t()
) :: State.t()
defp send_reply_for_unknown_enumeration(
source_address,
bvlc,
%NPCI{} = npci,
raw_apdu,
portal,
%State{} = state
) do
case APDU.get_invoke_id_from_raw_apdu(raw_apdu) do
{:ok, invoke_id} ->
device_id =
case npci.source do
%NpciTarget{address: adr} when adr != nil -> adr
_else -> nil
end
reject = %APDU.Reject{
invoke_id: invoke_id,
reason: Constants.macro_assert_name(:reject_reason, :undefined_enumeration)
}
case send_data(
reject,
get_reply_opts_for_npci(reject, bvlc, npci, state),
source_address,
portal,
device_id,
false,
nil,
state,
true
) do
{:ok, new_state} ->
new_state
{:error, err} ->
log_debug(fn ->
"Client: Encountered error on sending Reject reply due to unknown enumeration, error: #{inspect(err)}"
end)
state
end
_else ->
log_debug(fn ->
"Client: Unable to extract invoke ID from invalid APDU (caused by unknown enumeration)"
end)
state
end
end
@spec supports_segmentation(Constants.segmentation() | term()) :: boolean()
defp supports_segmentation(Constants.macro_assert_name(:segmentation, :segmented_both)),
do: true
defp supports_segmentation(Constants.macro_assert_name(:segmentation, :segmented_receive)),
do: true
defp supports_segmentation(_term), do: false
@spec get_reply_opts_for_npci(Protocol.apdu(), Protocol.bvlc(), NPCI.t(), State.t()) ::
Keyword.t()
defp get_reply_opts_for_npci(_apdu, _bvlc, %NPCI{} = npci, %State{} = _state) do
# Source gets always set in send_data/8
kw_put_lazy([], :destination, get_reply_opts_destination(npci))
end
defp get_reply_opts_destination(%NPCI{source: %NpciTarget{} = src} = _npci) do
src
end
defp get_reply_opts_destination(%NPCI{} = _npci) do
nil
end
@spec kw_put_lazy(Keyword.t(), atom(), term()) :: Keyword.t()
defp kw_put_lazy(kw, _key, nil), do: kw
defp kw_put_lazy(kw, key, val), do: Keyword.put(kw, key, val)
@spec kw_put_new(Keyword.t(), atom(), term() | nil) :: Keyword.t()
defp kw_put_new(kw, _key, nil), do: kw
defp kw_put_new(kw, key, val), do: Keyword.put_new(kw, key, val)
defp validate_start_link_opts(opts) do
case opts[:apdu_retries] do
nil ->
:ok
term when is_integer(term) and term >= 0 ->
:ok
term ->
raise ArgumentError,
"start_link/1 expected apdu_retries to be a non negative integer, " <>
"got: #{inspect(term)}"
end
case opts[:apdu_timeout] do
nil ->
:ok
term when is_integer(term) and term > 0 ->
:ok
term ->
raise ArgumentError,
"start_link/1 expected apdu_timeout to be a positive integer, " <>
"got: #{inspect(term)}"
end
case opts[:disable_app_timeout] do
nil ->
:ok
term when is_boolean(term) ->
:ok
term ->
raise ArgumentError,
"start_link/1 expected disable_app_timeout to be a boolean, " <>
"got: #{inspect(term)}"
end
case opts[:disable_invoke_id_management] do
nil ->
:ok
term when is_boolean(term) ->
:ok
term ->
raise ArgumentError,
"start_link/1 expected disable_invoke_id_management to be a boolean, " <>
"got: #{inspect(term)}"
end
case opts[:notification_receiver] do
nil ->
:ok
term when is_atom(term) ->
:ok
term when is_pid(term) ->
:ok
term when is_port(term) ->
:ok
term when is_tuple(term) and tuple_size(term) == 2 ->
:ok
term when is_list(term) ->
unless Enum.all?(
term,
&(is_atom(&1) or is_pid(&1) or is_port(&1) or
(is_tuple(&1) and tuple_size(&1) == 2))
) do
raise ArgumentError,
"start_link/1 expected notification_receiver to be a Process destination or " <>
"list of Process destinations, got: #{inspect(term)}"
end
term ->
raise ArgumentError,
"start_link/1 expected notification_receiver to be a Process destination or " <>
"list of Process destinations, got: #{inspect(term)}"
end
case opts[:npci_source] do
nil ->
:ok
%NpciTarget{} ->
:ok
term ->
raise ArgumentError,
"start_link/1 expected npci_source to be a NpciTarget, " <>
"got: #{inspect(term)}"
end
case opts[:segmentator] do
term when is_atom(term) ->
:ok
term when is_pid(term) ->
:ok
term when is_tuple(term) and tuple_size(term) == 2 ->
:ok
term when is_tuple(term) and tuple_size(term) == 3 ->
:ok
term ->
raise ArgumentError,
"start_link/1 expected segmentator to be a GenServer name, " <>
"got: #{inspect(term)}"
end
case opts[:segments_store] do
term when is_atom(term) ->
:ok
term when is_pid(term) ->
:ok
term when is_tuple(term) and tuple_size(term) == 2 ->
:ok
term when is_tuple(term) and tuple_size(term) == 3 ->
:ok
term ->
raise ArgumentError,
"start_link/1 expected segments_store to be a GenServer name, " <>
"got: #{inspect(term)}"
end
case opts[:segmented_rcv_window_overwrite] do
nil ->
:ok
term when is_boolean(term) ->
:ok
term ->
raise ArgumentError,
"start_link/1 expected segmented_rcv_window_overwrite to be a boolean, " <>
"got: #{inspect(term)}"
end
transport = opts[:transport]
case transport do
mod when is_atom(mod) ->
:ok
{mod, term} when is_atom(mod) and is_atom(term) ->
:ok
{mod, term} when is_atom(mod) and is_pid(term) ->
:ok
{mod, term} when is_atom(mod) and is_port(term) ->
:ok
{mod, term} when is_atom(mod) and is_tuple(term) and tuple_size(term) == 2 ->
:ok
{mod, term} when is_atom(mod) and is_tuple(term) and tuple_size(term) == 3 ->
:ok
term ->
raise ArgumentError,
"start_link/1 expected transport to be a tuple of module name " <>
"and TransportBehaviour.transport(), got: #{inspect(term)}"
end
# Unwrap the transport module name
{transport_mod, _pid} =
case transport do
{mod, pid} -> {mod, pid}
mod -> {mod, nil}
end
unless Code.ensure_loaded?(transport_mod) do
raise ArgumentError, "Given transport module #{inspect(transport_mod)} is not loaded"
end
unless Enum.any?(transport_mod.__info__(:attributes), fn
{:behaviour, TransportBehaviour} -> true
{:behaviour, [TransportBehaviour]} -> true
_else -> false
end) do
raise ArgumentError,
"Given transport module #{inspect(transport_mod)} does not " <>
"implement the BACnet transport behaviour"
end
end
defp validate_send_opts(opts, function) when is_list(opts) and is_binary(function) do
# Validate send options
Enum.each(opts, fn
{:device_id, device_id} ->
unless (is_integer(device_id) and device_id in 1..4_194_302) or is_nil(device_id) do
raise ArgumentError,
message:
"#{function} expected device_id in opts to be nil " <>
" or an integer in the range 1 - 4_194_302 inclusive, got: " <> inspect(device_id)
end
{:max_apdu_length, max_apdu_length} ->
unless is_integer(max_apdu_length) and max_apdu_length > 0 do
raise ArgumentError,
message:
"#{function} expected max_apdu_length in opts to be " <>
"a positive integer, got: " <> inspect(max_apdu_length)
end
{:max_segments, max_segments} ->
unless is_integer(max_segments) and max_segments > 0 do
raise ArgumentError,
message:
"#{function} expected max_segments in opts to be " <>
"a positive integer, got: " <> inspect(max_segments)
end
{:segmentation_supported, segmentation_supported} ->
unless Constants.has_by_name(:segmentation, segmentation_supported) do
raise ArgumentError,
message:
"#{function} expected segmentation_supported in opts to be " <>
"a valid atom, got: " <> inspect(segmentation_supported)
end
# Unknown option - ignore
_else ->
nil
end)
end
end