protocol.ex

defmodule Core.SyslogParser do
  require Logger

  alias Core.SyslogService

  def start_link(ref, socket, transport, opts) do
    pid = spawn_link(__MODULE__, :init, [ref, socket, transport, opts])
    {:ok, pid}
  end

  def init(ref, socket, transport, _Opts = []) do
    :ok = :ranch.accept_ack(ref)
    loop(socket, transport, "")
  end

  #The syslog protocol terminates messages with \n
  #That is why we can use the split function below to check if we have a whole
  #message or not. If we do actually have a whole message and no more it will end up
  #as a list of ["<123> ... syslogstuff", ""], meaning that its always safe to
  #send the remainder into the loop as a buffer part.
  def loop(socket, transport, buffer) do
    case transport.recv(socket, 0, 5000) do
      {:ok, data} ->
        rest = take_syslog_lines((buffer <> data))
        loop(socket, transport, rest)
      _ ->
        :ok = transport.close(socket)
    end
  end

  defp take_syslog_lines(data) do
    String.split(data, "\n", parts: 2)
    |> case do
      [rest] ->
        rest
      [message, rest] ->
        filter_and_process(message)
        take_syslog_lines(rest)
    end
  end

  defp filter_and_process(line) do
    # Logger.debug "Handling line:\n#{line}"
    pattern = :binary.compile_pattern([
                                        "Message=\"Conference has been created.\"",
                                        "Message=\"Participant has joined.\"",
                                        "Message=\"Participant has disconnected.\"",
                                        "Message=\"Media Stream destroyed\"",
                                        "Message=\"Conference has been stopped.\"",
                                        ])
    parse_message(line, String.contains?(line, pattern))
  end

  defp parse_message(_line, false), do: :noop

  defp parse_message(line, true) do
    # Logger.debug "Got filtered line to handle:\n#{line}"
    get_params(line, %{})
    |> SyslogService.handle_event
  end


  defp get_params("Service-tag=\"" <> rest, acc) do
    {rest, value} = get_quoted_value(rest, [])
    get_params(rest, Map.put(acc, :service_tag, value))
  end

  defp get_params("Conversation-id=\"" <> rest, acc) do
    {rest, value} = get_quoted_value(rest, [])
    get_params(rest, Map.put(acc, :conversation_id, value))
  end

  defp get_params("Participant=\"" <> rest, acc) do
    {rest, value} = get_quoted_value(rest, [])
    get_params(rest, Map.put(acc, :participant, value))
  end

  defp get_params("DisplayName=\"" <> rest, acc) do
    {rest, value} = get_quoted_value(rest, [])
    get_params(rest, Map.put(acc, :display_name, value))
  end

  defp get_params("Message=\"" <> rest, acc) do
    {rest, value} = get_quoted_value(rest, [])
    get_params(rest, Map.put(acc, :message, value))
  end

  defp get_params("Protocol=\"" <> rest, acc) do
    {rest, value} = get_quoted_value(rest, [])
    get_params(rest, Map.put(acc, :protocol, value))
  end

  defp get_params("Detail=\"" <> rest, acc) do
    {rest, value} = get_quoted_value(rest, [])
    get_params(rest, Map.put(acc, :detail, value))
  end

  defp get_params("", acc) do
    acc
  end

  #discard the rest as noise
  defp get_params(<<_::binary-size(1)>> <> rest, acc) do
    get_params(rest, acc)
  end



  def get_quoted_value("\"" <> rest, acc) do
    {rest, Enum.reverse(acc) |> Enum.join}
  end

  def get_quoted_value(<<x::binary-size(1)>> <> tail, acc) do
    get_quoted_value(tail, [x | acc])
  end
end

supervised_as_worker.ex

defmodule Core.SyslogListener do

  require Logger

  @port Application.get_env(:core, :syslog_port, 4001)
  def start_link do
    opts = [port: @port]
    Logger.debug "Setting up syslog listener at port #{@port}"
    {:ok, _} = :ranch.start_listener(:syslog_listener, 100, :ranch_tcp, opts, Core.SyslogParser, [])
  end
end