faktory_worker_ex

defmodule Faktory.Protocol do
  @moduledoc false

  # So we can use our send/2 defined below.
  import Kernel, except: [send: 2]

  def push(conn, job) when is_list(job), do: push(conn, Map.new(job))

  def push(conn, job) do
    payload = Poison.encode!(job)

    with :ok <- send(conn, "PUSH #{payload}"),
      {:ok, "+OK"} <- recv(conn, :line)
    do
      job["jid"]
    end
  end

  def fetch(conn, queues) when is_list(queues) do
    fetch(conn, Enum.join(queues, " "))
  end

  def fetch(conn, queues) when is_binary(queues) do
    with :ok <- send(conn, "FETCH #{queues}"),
      {:ok, <<"$", size::binary>>} <- recv(conn, :line),
      {:size, size} when size != "-1" <- {:size, size},
      {:ok, json} <- recv(conn, String.to_integer(size)),
      {:ok, ""} <- recv(conn, :line)
    do
      Poison.decode!(json)
    else
      {:size, "-1"} -> nil
      error -> error
    end
  end

  def ack(conn, jid) when is_binary(jid) do
    payload = %{"jid" => jid} |> Poison.encode!

    with :ok <- send(conn, "ACK #{payload}"),
      {:ok, "+OK"} <- recv(conn, :line)
    do
      {:ok, jid}
    end
  end

  def fail(conn, jid, errtype, message, backtrace) do
    payload = %{
      jid: jid,
      errtype: errtype,
      message: message,
      backtrace: backtrace
    } |> Poison.encode!

    with :ok <- send(conn, "FAIL #{payload}"),
      {:ok, "+OK"} <- recv(conn, :line)
    do
      {:ok, jid}
    end
  end

  def info(conn) do
    with :ok <- send(conn, "INFO"),
      {:ok, <<"$", size::binary>>} <- recv(conn, :line),
      size = String.to_integer(size),
      {:ok, json} <- recv(conn, size),
      {:ok, ""} <- recv(conn, :line)
    do
      Poison.decode(json)
    end
  end

  def beat(conn, wid) do
    payload = %{wid: wid} |> Poison.encode!

    with :ok <- send(conn, "BEAT #{payload}"),
      {:ok, "+OK"} <- recv(conn, :line)
    do
      :ok
    else
      {:ok, json} -> {:ok, Poison.decode!(json)}
      error -> error
    end
  end

  def flush(conn) do
    with :ok <- send(conn, "FLUSH"),
      {:ok, "+OK"} <- recv(conn, :line)
    do
      :ok
    end
  end

  defp send(conn, data) do
    Faktory.Connection.send(conn, data)
  end

  defp recv(conn, :line) do
    case Faktory.Connection.recv(conn, :line) do
      {:ok, <<"-ERR ", reason::binary>>} -> {:error, reason}
      {:ok, <<"-SHUTDOWN ", reason::binary>>} -> {:error, reason}
      {:ok, line} -> {:ok, line}
      error -> error
    end
  end

  defp recv(conn, size) do
    Faktory.Connection.recv(conn, size)
  end

end