defmodule FLHook.Client do @moduledoc """ A client to connect to a FLHook socket. """ use Connection require Logger alias __MODULE__ alias FLHook.Client.Reply alias FLHook.Codec alias FLHook.Command alias FLHook.CommandError alias FLHook.CommandSerializer alias FLHook.Config alias FLHook.ConfigError alias FLHook.Event alias FLHook.HandshakeError alias FLHook.Result alias FLHook.SocketError alias FLHook.Utils @welcome_msg "Welcome to FLHack" @typedoc """ Type representing a FLHook client process. """ @type client :: GenServer.server() defmacro __using__(opts) do otp_app = Keyword.fetch!(opts, :otp_app) quote location: :keep do @doc false @spec __config__() :: Config.t() def __config__ do unquote(otp_app) |> Application.get_env(__MODULE__, []) |> Config.new() end @doc """ Starts the FLHook client. """ @spec start_link(Keyword.t()) :: GenServer.on_start() def start_link(opts \\ []) do opts = Keyword.put_new(opts, :name, __MODULE__) Client.start_link(__config__(), opts) end @doc """ Default child specification for the FLHook client. """ @spec child_spec(term) :: Supervisor.child_spec() def child_spec(opts) do %{ id: __MODULE__, start: {__MODULE__, :start_link, [opts]} } end defoverridable child_spec: 1 end end @doc """ Starts the FLHook client using the given config or options. """ @spec start_link(Config.t() | Keyword.t()) :: GenServer.on_start() def start_link(%Config{} = config) do start_link(config, []) end def start_link(opts) when is_list(opts) do {config_opts, start_opts} = Keyword.split(opts, Map.keys(Config.__struct__())) config = Config.new(config_opts) start_link(config, start_opts) end @doc """ Starts the FLHook client using the given config and options. """ @spec start_link(Config.t(), GenServer.options()) :: GenServer.on_start() def start_link(%Config{} = config, opts) do Connection.start_link(__MODULE__, config, opts) end @doc """ Opens the connection. """ @spec open(client) :: :ok | {:error, Exception.t()} def open(client) do Connection.call(client, :open) end @doc """ Closes the connection. """ @spec close(client) :: :ok def close(client) do Connection.call(client, :close) end @doc """ Indicates whether the socket is connected. """ @spec connected?(client) :: boolean def connected?(client) do Connection.call(client, :connected?) end @doc false @spec cmd(client, Command.command()) :: {:ok, Result.t()} | {:error, Exception.t()} def cmd(client, cmd) do Connection.call(client, {:cmd, CommandSerializer.to_string(cmd)}) end @doc false @spec cmd!(client, Command.command()) :: Result.t() | no_return def cmd!(client, cmd) do case cmd(client, cmd) do {:ok, result} -> result {:error, error} -> raise error end end @doc false @spec subscribe(client, pid) :: :ok def subscribe(client, listener \\ self()) do Connection.call(client, {:subscribe, listener}) end @doc false @spec unsubscribe(client, pid) :: :ok def unsubscribe(client, listener \\ self()) do Connection.call(client, {:unsubscribe, listener}) end # Child Spec @doc """ The child specification for a FLHook client. """ @spec child_spec(term) :: Supervisor.child_spec() def child_spec(opts) do %{ id: __MODULE__, start: {__MODULE__, :start_link, [opts]} } end # Callbacks @impl true def init(%{password: nil}) do {:stop, %ConfigError{message: "No password specified"}} end def init(config) do state = %{ config: config, queue: nil, recv_timeout_ref: nil, socket: nil, listeners: %{} } if config.connect_on_start do {:connect, :init, state} else {:ok, state} end end @impl true def connect(info, %{config: config} = state) do case socket_connect(config) do {:ok, socket} -> with :ok <- verify_welcome_msg(socket, config), :ok <- authenticate(socket, config), :ok <- event_mode(socket, config) do :ok = config.tcp_adapter.controlling_process(socket, self()) config.inet_adapter.setopts(socket, active: :once) with {:open, from} <- info do Connection.reply(from, :ok) end {:ok, %{state | queue: :queue.new(), socket: socket}} else {:error, error} -> config.tcp_adapter.close(socket) state = %{state | socket: nil} case info do {:open, from} -> Connection.reply(from, {:error, error}) {:ok, state} _ -> log_error(error, config) {:stop, error, state} end end {:error, error} -> case info do {:open, from} -> Connection.reply(from, {:error, error}) {:ok, state} _ -> log_error(error, config) {:backoff, config.backoff_interval, state} end end end @impl true def disconnect(info, state) do maybe_cancel_timer(state.recv_timeout_ref) if state.socket do :ok = state.config.tcp_adapter.close(state.socket) end state = %{state | socket: nil} case info do {:close, from} -> Connection.reply(from, :ok) {:noconnect, state} error -> log_error(error, state.config) {:connect, :reconnect, state} end end @impl true def terminate(_reason, state) do maybe_cancel_timer(state.recv_timeout_ref) if state.socket do :ok = state.config.tcp_adapter.close(state.socket) end end @impl true def handle_call(:connected?, _from, state) do {:reply, !is_nil(state.socket), state} end def handle_call(:open, from, state) do {:connect, {:open, from}, state} end def handle_call(:close, from, state) do {:disconnect, {:close, from}, state} end def handle_call({:cmd, _cmd}, _from, %{socket: nil} = state) do {:reply, {:error, %SocketError{reason: :closed}}, state} end def handle_call({:cmd, cmd}, from, state) do case send_cmd(state.socket, state.config, cmd) do :ok -> queue = :queue.in(%Reply{client: from}, state.queue) recv_timeout_ref = Process.send_after(self(), :timeout, state.config.recv_timeout) {:noreply, %{state | queue: queue, recv_timeout_ref: recv_timeout_ref}} error -> {:reply, error, state} end end def handle_call({:subscribe, listener}, _from, state) do listeners = Map.put_new_lazy(state.listeners, listener, fn -> Process.monitor(listener) end) {:reply, :ok, %{state | listeners: listeners}} end def handle_call({:unsubscribe, listener}, _from, state) do listeners = case Map.pop(state.listeners, listener) do {nil, listeners} -> listeners {monitor_ref, listeners} -> Process.demonitor(monitor_ref) listeners end {:reply, :ok, %{state | listeners: listeners}} end @impl true def handle_info({:tcp, socket, msg}, state) do state.config.inet_adapter.setopts(socket, active: :once) case Codec.decode(state.config.codec, msg) do {:ok, msg} -> case Event.parse(msg) do {:ok, event} -> handle_event(event, state) :error -> handle_cmd_resp(msg, state) end {:error, error} -> log_error(error, state.config) {:stop, error, state} end end def handle_info({:tcp_closed, _socket}, state) do maybe_cancel_timer(state.recv_timeout_ref) {:disconnect, %SocketError{reason: :closed}, state} end def handle_info({:tcp_error, _socket, reason}, state) do maybe_cancel_timer(state.recv_timeout_ref) {:disconnect, %SocketError{reason: reason}, state} end def handle_info(:timeout, state) do maybe_cancel_timer(state.recv_timeout_ref) case :queue.out(state.queue) do {{:value, reply}, new_queue} -> Connection.reply( reply.client, {:error, %SocketError{reason: :timeout}} ) {:noreply, %{state | queue: new_queue, recv_timeout_ref: nil}} {:empty, _} -> {:noreply, state} end end def handle_info({:DOWN, monitor_ref, :process, listener, _info}, state) do {^monitor_ref, listeners} = Map.pop(state.listeners, listener) {:noreply, %{state | listeners: listeners}} end defp handle_event(event, state) do Enum.each(state.listeners, fn {listener, _monitor} -> send(listener, event) end) {:noreply, state} end defp handle_cmd_resp(msg, state) do maybe_cancel_timer(state.recv_timeout_ref) case :queue.out(state.queue) do {{:value, reply}, new_queue} -> case Reply.add_chunk(reply, msg) do %{status: :pending} = reply -> {:noreply, %{state | queue: :queue.in_r(reply, new_queue)}} %{status: :ok} = reply -> Connection.reply( reply.client, {:ok, Reply.to_result(reply)} ) {:noreply, %{state | queue: new_queue}} %{status: {:error, detail}} -> Connection.reply( reply.client, {:error, %CommandError{detail: detail}} ) {:noreply, %{state | queue: new_queue}} end {:empty, _queue} -> # should never happen {:noreply, state} end end # Helpers defp socket_connect(config) do with {:error, reason} <- config.tcp_adapter.connect( to_charlist(config.host), config.port, [:binary, active: false, send_timeout: config.send_timeout], config.connect_timeout ) do {:error, %SocketError{reason: reason}} end end defp read_chunk(socket, config) do with {:socket, {:ok, value}} <- {:socket, config.tcp_adapter.recv(socket, 0, config.recv_timeout)}, {:codec, {:ok, decoded}} <- {:codec, Codec.decode(config.codec, value)} do {:ok, decoded} else {:codec, error} -> error {:socket, {:error, reason}} -> {:error, %SocketError{reason: reason}} end end defp read_cmd_result(socket, config) do case do_read_cmd_result(%Reply{}, socket, config) do %Reply{status: :ok} = reply -> {:ok, Reply.to_result(reply)} %Reply{status: {:error, detail}} -> {:error, %CommandError{detail: detail}} error -> error end end defp do_read_cmd_result(%Reply{status: :pending} = reply, socket, config) do with {:ok, chunk} <- read_chunk(socket, config) do reply |> Reply.add_chunk(chunk) |> do_read_cmd_result(socket, config) end end defp do_read_cmd_result(%Reply{} = reply, _socket, _config), do: reply defp send_msg(socket, config, value) do with {:codec, {:ok, encoded}} <- {:codec, Codec.encode(config.codec, value)}, {:socket, :ok} <- {:socket, config.tcp_adapter.send(socket, encoded)} do :ok else {:codec, error} -> error {:socket, {:error, reason}} -> {:error, %SocketError{reason: reason}} end end defp send_cmd(socket, config, cmd) do send_msg(socket, config, cmd <> Utils.line_sep()) end defp cmd_passive(socket, config, cmd) do with :ok <- send_cmd(socket, config, CommandSerializer.to_string(cmd)) do read_cmd_result(socket, config) end end defp verify_welcome_msg(socket, config) do case read_chunk(socket, config) do {:ok, @welcome_msg <> _} -> :ok {:ok, message} -> {:error, %HandshakeError{actual_message: message}} error -> error end end defp authenticate(socket, config) do with {:ok, _} <- cmd_passive(socket, config, {"pass", [config.password]}) do :ok end end defp event_mode(_socket, %Config{event_mode: false}), do: :ok defp event_mode(socket, config) do with {:ok, _} <- cmd_passive(socket, config, "eventmode") do :ok end end defp maybe_cancel_timer(nil), do: :ok defp maybe_cancel_timer(timer_ref), do: Process.cancel_timer(timer_ref) defp log_error(error, config) do Logger.error( "FLHook (#{config.host}:#{config.port}): #{Exception.message(error)}" ) end end