nasa.ex

alias Experimental.GenStage

defmodule Go do
  def go do
    :ets.new(:logs, [:named_table, :public])


    {:ok, producer} = GenStage.from_enumerable(File.stream!("nasa.log"), name: __MODULE__)
    {:ok, consumer1} = GsConsumer.start_link("1")
    {:ok, consumer2} = GsConsumer.start_link("2")
    {:ok, consumer3} = GsConsumer.start_link("3")
    {:ok, consumer4} = GsConsumer.start_link("4")
  end

end

defmodule GsConsumer do
  use GenStage

  def start_link(num) do
    GenStage.start_link(__MODULE__, [num])
  end

  def init([num]) do
    {:consumer, num, subscribe_to: [Go]}
  end

  def handle_events(lines, _from, num) do
      lines
      |> Enum.map(fn(line) ->
        reg = ~r/GET \/[a-z]{0,10}\b([-a-zA-Z0-9@:%_\+.~#?&\/]*)/
        case Regex.run(reg, line) do
          nil -> ""
          [h | tail] -> String.to_atom(h)
        end
      end)
      |> Enum.each(fn(x) ->
          :ets.update_counter(:logs, x, 1, {x, 1})
      end)
    {:noreply, [], num}
  end
end