elixir_nsq

defmodule NSQ.Lookupd do
  alias NSQ.Connection, as: C
  require Logger

  @typedoc """
  All lookupd responses should return a map with these values. If the response
  is not of that form, it should be normalized into that form.
  """
  @type response :: %{
    data: binary,
    headers: [any],
    status_code: integer,
    status_txt: binary,
  }


  @spec nsqds_with_topic([C.host_with_port], String.t) :: [C.host_with_port]
  def nsqds_with_topic(lookupds, topic) do
    responses = Enum.map(lookupds, &topics_from_lookupd(&1, topic))
    nsqds = Enum.map responses, fn(response) ->
      Enum.map response["producers"] || [], fn(producer) ->
        if producer do
          {producer["broadcast_address"], producer["tcp_port"]}
        else
          nil
        end
      end
    end
    nsqds |>
      List.flatten |>
      Enum.uniq |>
      Enum.reject(fn(v) -> v == nil end)
  end


  @spec topics_from_lookupd(C.host_with_port, String.t) :: response
  def topics_from_lookupd({host, port}, topic) do
    lookupd_url = "http://#{host}:#{port}/lookup?topic=#{topic}"
    headers = [{"Accept", "application/vnd.nsq; version=1.0"}]
    try do
      case HTTPotion.get(lookupd_url, headers: headers) do
        %HTTPotion.Response{status_code: 200, body: body, headers: headers} ->
          normalize_200_response(headers, body)
        %HTTPotion.Response{status_code: 404} ->
          %{} |> normalize_response
        %HTTPotion.Response{status_code: status, body: body} ->
          Logger.error "Unexpected status code from #{lookupd_url}: #{status}"
          %{status_code: status, data: body}
          |> normalize_response
      end
    rescue
      e in HTTPotion.HTTPError ->
        Logger.error "Error connecting to #{lookupd_url}: #{inspect e}"
        %{status_code: nil, status_txt: nil, data: nil}
    end
  end


  @spec normalize_200_response([any], binary) :: response
  defp normalize_200_response(headers, body) do
    body = if body == nil || body == "", do: "{}", else: body

    if headers[:"X-Nsq-Content-Type"] == "nsq; version=1.0" do
      Poison.decode!(body)
      |> normalize_response
    else
      %{status_code: 200, status_txt: "OK", data: body}
      |> normalize_response
    end
  end


  @spec normalize_response(map) :: response
  defp normalize_response(m) do
    Map.merge(%{
      status_code: nil,
      status_txt: "",
      data: "",
      headers: []
    }, m)
  end
end