Packages

verk_web

1.5.1
1.6.0 1.5.1 1.5.0 1.4.2 1.4.1 1.4.0 1.3.0 1.2.0 1.1.1 1.1.0 1.0.2 1.0.1 1.0.0 0.14.0 0.13.7 0.13.6 0.13.5 0.13.4 0.13.3 retired 0.13.2 0.13.1 0.13.0 0.11.0 0.10.0 0.9.5 0.9.4 0.9.3 0.9.2 0.9.1 0.9.0

A Verk dashboard

HexDocs

Current section

Files

Jump to
Readme
30 Versions
10 Dependencies
Dependants
Files
Activity
verk_web lib verk_web tracking_jobs_handler.ex
Raw

lib/verk_web/tracking_jobs_handler.ex 

defmodule VerkWeb.TrackingJobsHandler do

  @broadcast_interval 1_000



  use GenStage



  def start_link(pid) do

    GenStage.start_link(__MODULE__, pid)

  end



  def init(pid) do

    Process.send_after(self(), :broadcast_stats, @broadcast_interval)

    {:consumer, {pid, %{finished: 0, failed: 0}}, subscribe_to: [Verk.EventProducer]}

  end



  def handle_events(events, _from, {pid, stats}) do

    stats = Enum.reduce(events, stats, &handle_event(&1, &2))

    {:noreply, [], {pid, stats}}

  end



  defp handle_event(%Verk.Events.JobFinished{}, stats) do

    %{finished: stats[:finished] + 1, failed: stats[:failed]}

  end



  defp handle_event(%Verk.Events.JobFailed{}, stats) do

    %{finished: stats[:finished], failed: stats[:failed] + 1}

  end



  defp handle_event(_, stats), do: stats



  def handle_info(:broadcast_stats, {pid, stats}) do

    send(pid, {:stats, stats})

    Process.send_after(self(), :broadcast_stats, @broadcast_interval)

    {:noreply, [], {pid, %{finished: 0, failed: 0}}}

  end



  def handle_info(_, state), do: {:noreply, [], state}

end