defmodule MqttX.Transport.ThousandIsland do @moduledoc """ ThousandIsland transport adapter for MqttX. This adapter uses ThousandIsland as the underlying TCP/TLS server. ## Usage MqttX.Server.start_link(MyHandler, handler_opts, transport: MqttX.Transport.ThousandIsland, port: 1883 ) ## Options - `:port` - Port to listen on (default: 1883) - `:ip` - IP address to bind to (default: `{0, 0, 0, 0}`) - `:transport_module` - ThousandIsland transport (`:tcp` or `:ssl`) - `:transport_options` - SSL/TLS options when using `:ssl` - `:num_acceptors` - Number of acceptor processes (default: 100) """ @behaviour MqttX.Transport require Logger @default_port 1883 @default_num_acceptors 100 @impl MqttX.Transport def start_link(handler, handler_opts, transport_opts) do port = Keyword.get(transport_opts, :port, @default_port) ip = Keyword.get(transport_opts, :ip, {0, 0, 0, 0}) num_acceptors = Keyword.get(transport_opts, :num_acceptors, @default_num_acceptors) transport_module = Keyword.get(transport_opts, :transport_module, ThousandIsland.Transports.TCP) transport_options = Keyword.get(transport_opts, :transport_options, []) # Create ETS table for retained messages retained_table = create_retained_table(port) # Create rate limiter if configured rate_limiter = case Keyword.get(transport_opts, :rate_limit) do nil -> nil rate_limit_opts -> MqttX.Server.RateLimiter.new(rate_limit_opts) end handler_module = __MODULE__.Handler handler_opts_full = %{ handler: handler, handler_opts: handler_opts, transport_opts: transport_opts, retained_table: retained_table, rate_limiter: rate_limiter } thousand_island_opts = [ port: port, handler_module: handler_module, handler_options: handler_opts_full, transport_module: transport_module, transport_options: [{:ip, ip} | transport_options], num_acceptors: num_acceptors ] Logger.info("[MqttX.Transport.ThousandIsland] Starting on port #{port}") ThousandIsland.start_link(thousand_island_opts) end defp create_retained_table(port) do table_name = :"mqttx_retained_#{port}" case :ets.whereis(table_name) do :undefined -> :ets.new(table_name, [:named_table, :public, :set]) _ref -> # Table already exists, return the name table_name end end @impl MqttX.Transport def send(socket, data) do ThousandIsland.Socket.send(socket, data) end @impl MqttX.Transport def close(socket) do ThousandIsland.Socket.close(socket) :ok end @impl MqttX.Transport def peername(socket) do ThousandIsland.Socket.peername(socket) end @impl MqttX.Transport def getopts(socket, opts) do ThousandIsland.Socket.getopts(socket, opts) end @impl MqttX.Transport def setopts(socket, opts) do ThousandIsland.Socket.setopts(socket, opts) end # Inner handler module that implements ThousandIsland.Handler defmodule Handler do @moduledoc false use ThousandIsland.Handler alias MqttX.Transport.Handler, as: Proto @impl ThousandIsland.Handler def handle_connection(socket, state) do send_fn = fn data -> ThousandIsland.Socket.send(socket, data) end case Proto.init( state.handler, state.handler_opts, state.retained_table, state.rate_limiter, send_fn ) do {:ok, proto} -> {:continue, proto} {:error, :rate_limited} -> {:close, state} end end @impl ThousandIsland.Handler def handle_data(data, _socket, state) do case Proto.handle_data(data, state) do {:ok, s} -> {:continue, s} {:close, _, s} -> {:close, s} {:error, _, s} -> {:close, s} end end @impl ThousandIsland.Handler def handle_close(_socket, state) do Proto.handle_close(state) end @impl ThousandIsland.Handler def handle_error(reason, _socket, state) do Proto.handle_error(reason, state) end @impl ThousandIsland.Handler def handle_timeout(_socket, state) do Proto.handle_timeout(state) end @impl GenServer def handle_info(msg, {socket, state}) do case Proto.handle_info(msg, state) do {:noreply, s} -> {:noreply, {socket, s}} {:stop, r, s} -> {:stop, r, {socket, s}} end end end end