safe_tcp.ex

defmodule SafeTcp do
  @moduledoc"""
  This process tries to maintain connection with the TCP server. If error is
  occured or connection is closed - it will be trying to establish the new one every
  @reconnect_freequency milliseconds
  """

  use GenServer

  @initial_state %{socket: nil, connection_error: nil, host: nil, port: nil}
  @reconnect_freequency 5 * 1000
  @timeout 1000

  @typep state :: %{
    socket: :gen_tcp.socket,
    connection_error: atom,
    host: bitstring,
    port: number
  }

  @spec start_link(%{host: bitstring, port: number}) :: GenServer.on_start
  def start_link(%{host: _, port: _} = server) do
    GenServer.start_link(__MODULE__, Map.merge(@initial_state, server))
  end

  @spec init(state) :: {:ok, state}
  def init(state) do
    send self(), :connect # don't block process from where it was called
    {:ok, state}
  end

  @spec send_event(pid, any) :: any
  def send_event(pid, event) do
    GenServer.call(pid, {:send, event})
  end

  @spec handle_call(any, GenServer.from, state) :: {:reply, any, state}
  def handle_call(
    {:send, _event}, _from, %{socket: nil, connection_error: error} = state
  ), do: {:reply, {:error, error}, state}

  def handle_call({:send, event}, _from, %{socket: socket} = state) do
    result = case :gen_tcp.send(socket, event) do
      {:error, reason} when reason in [:closed, :enotconn] ->
        trigger_reconnect()
        {:error, reason}
      other -> other
    end
    {:reply, result, state}
  end

  @spec handle_info(any, state) :: {:noreply, state}
  def handle_info({:tcp_closed, _}, state), do: on_error(state, :tcp_closed)
  def handle_info({:tcp_error, _, reason}, state), do: on_error(state, reason)

  def handle_info(:connect, %{host: host, port: port} = state) do
    opts = [:binary, active: true]
    case :gen_tcp.connect(host, port, opts, @timeout) do
      {:ok, socket} ->
        {:noreply, %{state | connection_error: nil, socket: socket}}
      {:error, error} ->
        trigger_reconnect()
        {:noreply, %{state | connection_error: String.to_atom("connection_#{error}")}}
    end
  end

  def handle_info({:tcp, _, _}, state), do: {:noreply, state}

  @spec on_error(state, atom) :: {:noreply, state}
  defp on_error(state, error) do
    trigger_reconnect()
    {:noreply, %{state | connection_error: error}}
  end

  @spec trigger_reconnect :: any
  defp trigger_reconnect do
    IO.puts("#{inspect(self())} reconnecting!")
    Process.send_after self(), :connect, @reconnect_freequency
  end
end