defmodule Tortoise.Handler do
@moduledoc """
User defined callback module for handling connection life cycle events.
`Tortoise.Handler` defines a behaviour which can be given to a
`Tortoise.Connection`. This allow the user to implement
functionality for events happening in the life cycle of the
connection, the provided callbacks are:
- `init/1` and `terminate/2` are called when the connection is
started and stopped. Parts that are setup in `init/1` can be
torn down in `terminate/2`. Notice that the connection process
is not terminated if the connection to the broker is lost;
`Tortoise` will attempt to reconnect, and events that should
happen when the connection goes offline should be set up using
the `connection/3` callback.
- `last_will/1` is called to get a last will message for a
new connection. If none is provided (nil), the connection's default
last will will be used.
- `connection/3` is called when the connection is `:up` or
`:down`, allowing for functionality to be run when the
connection state changes.
- `subscription/3` is called when a topic filter subscription
changes status, so this callback can be used to control the
life-cycle of a subscription, allowing us to implement custom
behavior for when the subscription is accepted, declined, as
well as unsubscribed.
- `handle_message/3` is run when the client receive a message on
one of the subscribed topic filters.
Because the callback-module will run inside the connection
controller process, which also handles the routing of protocol
messages (such as publish acknowledge messages) it is important that
the callbacks do not call functions that will block the process,
especially for clients that subscribe to topics with heavy
traffic.
Technically it would be possible to run the callback module in a
different process than the controller process, but it has been
decided to keep it on the controller as we otherwise would have to
copy every message evaluated by the connection to another
process. It is much better to let the end-user handle the
dispatching to other parts of the system while we are evaluating
what to do with the process anyways with the caveats:
- The callbacks should not block the controller
- it is not possible to call the (un)subscribe and publish
functions from within a callback as they will block the
controller.
While it is not possible to subscribe and unsubscribe in the handler
process using the `Tortoise.subscribe/3` and
`Tortoise.unsubscribe/3` it is possible to make changes to the
subscription list via `:gen_statem` inspired *next_actions*.
## Next actions
In some situations one would like to subscribe or unsubscribe a
filter topic when a certain event happens on the client. The
functions for interacting with the MQTT broker, defined on the
`Tortoise`-module are for the most part blocking operations, or
require the user to peek into the process mailbox to fetch the
result of the operation. To allow for changes in the subscriptions
one can define a set of next actions that should happen as part of
the return value to the `handle_message/3`, `subscription/3`, and
`connection/3` callbacks by returning a `{:ok, state, next_actions}`
where `next_actions` is a list of commands of:
- `{:subscribe, topic_filter, qos: qos, timeout: 5000}` where
`topic_filter` is a binary containing a valid MQTT topic filter,
and `qos` is the desired quality of service (0..2). The timeout
is the amount of time in milliseconds we are willing to wait for
a response to the request.
- `{:unsubscribe, topic_filter}` where `topic_filter` is a binary
containing the name of the subscription we want to unsubscribe
from.
If we want to unsubscribe from the current topic when we receive a
message on it we could write a `handle_message/3` as follows:
def handle_message(topic, _payload, state) do
topic = Enum.join(topic, "/")
next_actions = [{:unsubscribe, topic}]
{:ok, state, next_actions}
end
Note that the `topic` is received as a list of topic levels, and
that the next actions has to be a list, even if there is only one
next action; multiple actions can be given at once. Read more about
this in the `handle_message/3` documentation.
"""
alias Tortoise.Package
@typedoc """
Data structure describing the user defined callback handler
The data structure describe the current state as well as its initial
arguments and the module driving the handler. This allow Tortoise to
restart the handler if needed be.
"""
@type t :: %__MODULE__{
module: module(),
initial_args: term(),
state: term()
}
@enforce_keys [:module, :initial_args]
defstruct module: nil, state: nil, initial_args: []
# Helper for building a Handler struct so we can keep it as an
# opaque type in the system.
@doc false
@spec new({module(), args :: term()}) :: t
def new({module, args}) when is_atom(module) and is_list(args) do
%__MODULE__{module: module, initial_args: args}
end
# identity
def new(%__MODULE__{} = handler), do: handler
defmacro __using__(_opts) do
quote location: :keep do
@behaviour Tortoise.Handler
@impl true
def init(state) do
{:ok, state}
end
@impl true
def terminate(_reason, _state) do
:ok
end
@impl true
def connection(_status, state) do
{:ok, state}
end
@impl true
def subscription(_status, _topic_filter, state) do
{:ok, state}
end
@impl true
def handle_message(_topic, _payload, state) do
{:ok, state}
end
@impl true
def last_will(state) do
{{:ok, nil}, state}
end
defoverridable Tortoise.Handler
end
end
@typedoc """
Action to perform before reentering the execution loop.
The supported next actions are:
- Tell the connection process to subscribe to a topic filter
- Tell the connection process to unsubscribe from a topic filter
More next actions might be supported in the future.
"""
@type next_action() ::
{:subscribe, Tortoise.topic_filter(), [{:qos, Tortoise.qos()} | {:timeout, timeout()}]}
| {:unsubscribe, Tortoise.topic_filter()}
@doc """
Invoked when the connection is started.
`args` is the argument passed in from the connection configuration.
Returning `{:ok, state}` will let the MQTT connection receive data
from the MQTT broker, and the value contained in `state` will be
used as the process state.
"""
@callback init(args :: term()) :: {:ok, state}
when state: any()
@doc """
Invoked when a new connection is being attempted to set the last will message to one
provided by the handler. If the handler returns {{:ok, nil}, state}, the pre-set
last will is used.
"""
@callback last_will(state) :: {{:ok, term() | nil}, state} when state: any()
@doc """
Invoked when the connection status changes.
`status` is one of `:up` or `:down`, where up means we have an open
connection to the MQTT broker, and down means the connection is
temporary down. The connection process will attempt to reestablish
the connection.
Returning `{:ok, new_state}` will set the state for later
invocations.
Returning `{:ok, new_state, next_actions}`, where `next_actions` is
a list of next actions such as `{:unsubscribe, "foo/bar"}` will
result in the state being returned and the next actions performed.
"""
@callback connection(status, state :: term()) ::
{:ok, new_state}
| {:ok, new_state, [next_action()]}
when status: :up | :down,
new_state: term()
@doc """
Invoked when the subscription of a topic filter changes status.
The `status` of a subscription can be one of:
- `:up`, triggered when the subscription has been accepted by the
MQTT broker with the requested quality of service
- `{:warn, [requested: req_qos, accepted: qos]}`, triggered when
the subscription is accepted by the MQTT broker, but with a
different quality of service `qos` than the one requested
`req_qos`
- `{:error, reason}`, triggered when the subscription is rejected
with the reason `reason` such as `:access_denied`
- `:down`, triggered when the subscription of the given topic
filter has been successfully acknowledged as unsubscribed by the
MQTT broker
The `topic_filter` is the topic filter in question, and the `state`
is the internal state being passed through transitions.
Returning `{:ok, new_state}` will set the state for later
invocations.
Returning `{:ok, new_state, next_actions}`, where `next_actions` is
a list of next actions such as `{:unsubscribe, "foo/bar"}` will
result in the state being returned and the next actions performed.
"""
@callback subscription(status, topic_filter, state :: term) ::
{:ok, new_state}
| {:ok, new_state, [next_action()]}
when status:
:up
| :down
| {:warn, [requested: Tortoise.qos(), accepted: Tortoise.qos()]}
| {:error, term()},
topic_filter: Tortoise.topic_filter(),
new_state: term
@doc """
Invoked when messages are published to subscribed topics.
The `topic` comes in the form of a list of binaries, making it
possible to pattern match on the topic levels of the retrieved
message, store the individual topic levels as variables and use it
in the function body.
`Payload` is a binary. MQTT 3.1.1 does not specify any format of the
payload, so it has to be decoded and validated depending on the
needs of the application.
In an example where we are already subscribed to the topic filter
`room/+/temp` and want to dispatch the received messages to a
`Temperature` application we could set up our `handle_message` as
such:
def handle_message(["room", room, "temp"], payload, state) do
:ok = Temperature.record(room, payload)
{:ok, state}
end
Notice; the `handle_message/3`-callback run inside the connection
controller process, so for handlers that are subscribing to topics
with heavy traffic should do as little as possible in the callback
handler and dispatch to other parts of the application using
non-blocking calls.
Returning `{:ok, new_state}` will reenter the loop and set the state
for later invocations.
Returning `{:ok, new_state, next_actions}`, where `next_actions` is
a list of next actions such as `{:unsubscribe, "foo/bar"}` will
reenter the loop and perform the listed actions.
"""
@callback handle_message(topic_levels, payload, state :: term()) ::
{:ok, new_state}
| {:ok, new_state, [next_action()]}
when new_state: term(),
topic_levels: [String.t()],
payload: Tortoise.payload()
@doc """
Invoked when the connection process is about to exit.
If anything is setup during the `init/1` callback it should get
cleaned up during the `terminate/2` callback.
"""
@callback terminate(reason, state :: term) :: ignored
when reason: :normal | :shutdown | {:shutdown, term()},
ignored: term()
@optional_callbacks last_will: 1
@doc false
@spec execute(t, action) ::
:ok | {:ok, t} | {{:ok, any()}, t} | {:error, {:invalid_next_action, term()}}
when action:
:init
| :last_will
| {:subscribe, [term()]}
| {:unsubscribe, [term()]}
| {:publish, Tortoise.Package.Publish.t()}
| {:connection, :up | :down}
| {:terminate, reason :: term()}
def execute(handler, :init) do
case apply(handler.module, :init, [handler.initial_args]) do
{:ok, initial_state} ->
{:ok, %__MODULE__{handler | state: initial_state}}
end
end
def execute(handler, :last_will) do
handler.module
|> apply(:last_will, [handler.state])
|> handle_result(handler)
end
def execute(handler, {:connection, status}) do
handler.module
|> apply(:connection, [status, handler.state])
|> handle_result(handler)
end
def execute(handler, {:publish, %Package.Publish{} = publish}) do
topic_list = String.split(publish.topic, "/")
handler.module
|> apply(:handle_message, [topic_list, publish.payload, handler.state])
|> handle_result(handler)
end
def execute(handler, {:unsubscribe, unsubacks}) do
Enum.reduce(unsubacks, {:ok, handler}, fn topic_filter, {:ok, handler} ->
handler.module
|> apply(:subscription, [:down, topic_filter, handler.state])
|> handle_result(handler)
# _, {:stop, acc} ->
# {:stop, acc}
end)
end
def execute(handler, {:subscribe, subacks}) do
subacks
|> flatten_subacks()
|> Enum.reduce({:ok, handler}, fn {op, topic_filter}, {:ok, handler} ->
handler.module
|> apply(:subscription, [op, topic_filter, handler.state])
|> handle_result(handler)
# _, {:stop, acc} ->
# {:stop, acc}
end)
end
def execute(handler, {:terminate, reason}) do
_ignored = apply(handler.module, :terminate, [reason, handler.state])
:ok
end
# Subacks will come in a map with three keys in the form of tuples
# where the fist element is one of `:ok`, `:warn`, or `:error`. This
# is done to make it easy to pattern match in other parts of the
# system, and error out early if the result set contain errors. In
# this part of the system it is more convenient to transform the
# data to a flat list containing tuples of `{operation, data}` so we
# can reduce the handler state to collect the possible next actions,
# and pass through if there is an :error or :disconnect return.
defp flatten_subacks(subacks) do
Enum.reduce(subacks, [], fn
{_, []}, acc ->
acc
{:ok, entries}, acc ->
for {topic_filter, _qos} <- entries do
{:up, topic_filter}
end ++ acc
{:warn, entries}, acc ->
for {topic_filter, warning} <- entries do
{{:warn, warning}, topic_filter}
end ++ acc
{:error, entries}, acc ->
for {reason, {topic_filter, _qos}} <- entries do
{{:error, reason}, topic_filter}
end ++ acc
end)
end
# handle the user defined return from the callback
defp handle_result({:ok, updated_state}, handler) do
{:ok, %__MODULE__{handler | state: updated_state}}
end
defp handle_result({{:ok, answer}, updated_state}, handler) do
{{:ok, answer}, %__MODULE__{handler | state: updated_state}}
end
defp handle_result({:ok, updated_state, next_actions}, handler)
when is_list(next_actions) do
case Enum.split_with(next_actions, &valid_next_action?/1) do
{next_actions, []} ->
# send the next actions to the process mailbox. Notice that
# this code is run in the context of the connection controller
for action <- next_actions, do: send(self(), {:next_action, action})
{:ok, %__MODULE__{handler | state: updated_state}}
{_, errors} ->
{:error, {:invalid_next_action, errors}}
end
end
defp valid_next_action?({:subscribe, topic, opts}) do
is_binary(topic) and is_list(opts)
end
defp valid_next_action?({:unsubscribe, topic}) do
is_binary(topic)
end
defp valid_next_action?(_otherwise), do: false
end
