redix

defmodule Redix.Connection.Receiver do
  @moduledoc false

  use GenServer

  alias Redix.Protocol

  @initial_state %{
    # The process that sends stuff to the socket and that spawns this process
    sender: nil,
    # The queue of commands issued to Redis
    queue: :queue.new,
    # The TCP socket, which should be passive when given to this process
    socket: nil,
    # The tail of unparsed data
    tail: "",
  }

  @doc """
  Starts this genserver.

  Options in `opts` are injected directly in the state of this genserver.
  """
  @spec start_link(Keyword.t) :: GenServer.on_start
  def start_link(opts) do
    GenServer.start_link(__MODULE__, opts)
  end

  @doc """
  Puts `what` in the internal queue of genserver `pid` asynchronously (cast).
  """
  @spec enqueue(pid, term) :: :ok
  def enqueue(pid, what) do
    GenServer.cast(pid, {:enqueue, what})
  end

  ## Callbacks

  @doc false
  def init(opts) do
    state = Dict.merge(@initial_state, opts)
    :inet.setopts(state.socket, active: :once)
    {:ok, state}
  end

  @doc false
  def handle_cast({:enqueue, what}, state) do
    state = update_in(state.queue, &:queue.in(what, &1))
    {:noreply, state}
  end

  @doc false
  def handle_info({:tcp, socket, data}, %{socket: socket} = state) do
    :ok = :inet.setopts(socket, active: :once)
    state = new_data(state, state.tail <> data)
    {:noreply, state}
  end

  def handle_info({:tcp_closed, socket} = msg, %{socket: socket} = state) do
    disconnect(msg, {:error, :disconnected}, state)
  end

  def handle_info({:tcp_error, socket, reason} = msg, %{socket: socket} = state) do
    disconnect(msg, {:error, reason}, state)
  end

  ## Helpers

  defp new_data(state, <<>>) do
    %{state | tail: <<>>}
  end

  defp new_data(state, data) do
    {{:value, {:commands, from, ncommands}}, new_queue} = :queue.out(state.queue)

    case Protocol.parse_multi(data, ncommands) do
      {:ok, resp, rest} ->
        Connection.reply(from, format_resp(resp))
        state = %{state | queue: new_queue}
        new_data(state, rest)
      {:error, :incomplete} ->
        %{state | tail: data}
    end
  end

  defp disconnect(msg, error, state) do
    state = reply_to_queue(error, state)
    send state.sender, {:receiver, self(), msg}
    {:stop, :normal, state}
  end

  defp reply_to_queue(error, state) do
    for {:commands, from, _} <- :queue.to_list(state.queue) do
      Connection.reply(from, error)
    end

    %{state | queue: :queue.new}
  end

  defp format_resp(%Redix.Error{} = err), do: {:error, err}
  defp format_resp(resp), do: {:ok, resp}
end