+
+ def maybe_start_job(running_jobs, queue) do
+ if (:sets.size(running_jobs) < @max_jobs) && !:queue.is_empty(queue) do
+ {{:value, {type, payload}}, queue} = :queue.out(queue)
+ {:ok, pid} = Task.start(fn -> handle(type, payload) end)
+ mref = Process.monitor(pid)
+ {:sets.add_element(mref, running_jobs), queue}
+ else
+ {running_jobs, queue}
+ end
+ end
+
+ def handle_cast({:enqueue, type, payload}, {running_jobs, queue}) do
+ queue = :queue.in({type, payload}, queue)
+ {running_jobs, queue} = maybe_start_job(running_jobs, queue)
+ {:noreply, {running_jobs, queue}}
+ end
+
+ def handle_info({:DOWN, ref, :process, _pid, _reason}, {running_jobs, queue}) do
+ running_jobs = :sets.del_element(ref, running_jobs)
+ {running_jobs, queue} = maybe_start_job(running_jobs, queue)
+ {:noreply, {running_jobs, queue}}
+ end
+
+ def handle_cast(m, state) do
+ IO.inspect("Unknown: #{inspect(m)}, #{inspect(state)}")
+ {:noreply, state}
+ end