kafka_ex

defmodule KafkaEx.NetworkClient do
  require Logger
  alias KafkaEx.Protocol.Metadata.Broker
  alias KafkaEx.Socket

  @moduledoc false
  @spec create_socket(binary, non_neg_integer, KafkaEx.ssl_options, boolean) :: nil | Socket.t
  def create_socket(host, port, ssl_options \\ [], use_ssl \\ false) do
    case Socket.create(format_host(host), port, build_socket_options(ssl_options), use_ssl) do
      {:ok, socket} ->
        Logger.log(:debug, "Succesfully connected to broker #{inspect(host)}:#{inspect port}")
        socket
      err           ->
        Logger.log(:error, "Could not connect to broker #{inspect(host)}:#{inspect port} because of error #{inspect err}")
        nil
    end
  end

  @spec close_socket(nil | Socket.t) :: :ok
  def close_socket(nil), do: :ok
  def close_socket(socket), do: Socket.close(socket)

  @spec send_async_request(Broker.t, iodata) :: :ok | {:error, :closed | :inet.posix}
  def send_async_request(broker, data) do
    socket = broker.socket
    case Socket.send(socket, data) do
      :ok -> :ok
      {_, reason} ->
        Logger.log(:error, "Asynchronously sending data to broker #{inspect broker.host}:#{inspect broker.port} failed with #{inspect reason}")
        reason
    end
  end

  @spec send_sync_request(Broker.t, iodata, timeout) :: nil | iodata
  def send_sync_request(%{:socket => socket} = broker, data, timeout) do
    :ok = Socket.setopts(socket, [:binary, {:packet, 4}, {:active, false}])
    response = case Socket.send(socket, data) do
      :ok ->
        case Socket.recv(socket, 0, timeout) do
          {:ok, data} -> data
          {:error, reason} ->
            Logger.log(:error, "Receiving data from broker #{inspect broker.host}:#{inspect broker.port} failed with #{inspect reason}")
            nil
        end
      {_, reason} ->
        Logger.log(:error, "Sending data to broker #{inspect broker.host}:#{inspect broker.port} failed with #{inspect reason}")
        nil
    end

    :ok = Socket.setopts(socket, [:binary, {:packet, 4}, {:active, true}])
    response
  end

  @spec format_host(binary) :: char_list | :inet.ip_address
  def format_host(host) do
    case Regex.scan(~r/^(\d{1,3})\.(\d{1,3})\.(\d{1,3})\.(\d{1,3})$/, host) do
      [match_data] = [[_, _, _, _, _]] -> match_data |> tl |> List.flatten |> Enum.map(&String.to_integer/1) |> List.to_tuple
      _ -> to_char_list(host)
    end
  end

  defp build_socket_options([]) do
    [:binary, {:packet, 4}]
  end
  defp build_socket_options(ssl_options) do
    build_socket_options([]) ++ ssl_options
  end
end