Packages
verk
1.7.1
1.7.3
1.7.2
1.7.1
1.7.0
1.6.4
1.6.3
1.6.2
1.6.1
1.6.0
1.5.0
1.4.1
1.4.0
1.3.0
1.2.4
1.2.3
1.2.2
1.2.1
1.2.0
1.1.1
1.1.0
1.0.2
1.0.1
1.0.0
0.14.1
0.14.0
0.13.6
0.13.5
0.13.4
0.13.3
0.13.2
0.13.1
0.13.0
0.12.1
0.12.0
0.11.1
0.11.0
0.10.0
0.9.13
0.9.12
0.9.11
0.9.10
0.9.9
0.9.8
0.9.7
0.9.6
0.9.5
0.9.4
0.9.3
0.9.2
0.9.1
0.9.0
Verk is a job processing system backed by Redis.
Current section
Files
Jump to
Current section
Files
lib/verk/event_producer.ex
defmodule Verk.EventProducer do
@moduledoc """
A GenStage producer that broadcasts events to subscribed consumers.
"""
use GenStage
def start_link do
GenStage.start_link(__MODULE__, :ok, name: __MODULE__)
end
def stop, do: GenStage.stop(__MODULE__)
def async_notify(event) do
GenStage.cast(__MODULE__, {:notify, event})
end
def init(:ok) do
{:producer, {:queue.new(), 0}, dispatcher: GenStage.BroadcastDispatcher}
end
def handle_cast({:notify, event}, {queue, demand}) do
dispatch_events(:queue.in(event, queue), demand)
end
def handle_demand(incoming_demand, {queue, demand}) do
dispatch_events(queue, incoming_demand + demand)
end
defp dispatch_events(queue, demand, events \\ []) do
with d when d > 0 <- demand,
{{:value, event}, queue} <- :queue.out(queue) do
dispatch_events(queue, demand - 1, [event | events])
else
_ -> {:noreply, Enum.reverse(events), {queue, demand}}
end
end
end