Commit 52aa65fe authored by lain's avatar lain
Browse files

Federator Queue: Add seperate in/out queues.

parent 5ff5d583
Pipeline #269 passed with stage
in 1 minute and 59 seconds
...@@ -14,7 +14,10 @@ def start_link do ...@@ -14,7 +14,10 @@ def start_link do
Process.sleep(1000 * 60 * 1) # 1 minute Process.sleep(1000 * 60 * 1) # 1 minute
enqueue(:refresh_subscriptions, nil) enqueue(:refresh_subscriptions, nil)
end) end)
GenServer.start_link(__MODULE__, {:sets.new(), :queue.new()}, name: __MODULE__) GenServer.start_link(__MODULE__, %{
in: {:sets.new(), :queue.new()},
out: {:sets.new(), :queue.new()}
}, name: __MODULE__)
end end
def handle(:refresh_subscriptions, _) do def handle(:refresh_subscriptions, _) do
...@@ -95,10 +98,18 @@ def maybe_start_job(running_jobs, queue) do ...@@ -95,10 +98,18 @@ def maybe_start_job(running_jobs, queue) do
end end
end end
def handle_cast({:enqueue, type, payload}, {running_jobs, queue}) do def handle_cast({:enqueue, type, payload}, state) when type in [:incoming_doc] do
queue = :queue.in({type, payload}, queue) %{in: {i_running_jobs, i_queue}, out: {o_running_jobs, o_queue}} = state
{running_jobs, queue} = maybe_start_job(running_jobs, queue) i_queue = :queue.in({type, payload}, i_queue)
{:noreply, {running_jobs, queue}} {i_running_jobs, i_queue} = maybe_start_job(i_running_jobs, i_queue)
{:noreply, %{in: {i_running_jobs, i_queue}, out: {o_running_jobs, o_queue}}}
end
def handle_cast({:enqueue, type, payload}, state) do
%{in: {i_running_jobs, i_queue}, out: {o_running_jobs, o_queue}} = state
o_queue = :queue.in({type, payload}, o_queue)
{o_running_jobs, o_queue} = maybe_start_job(o_running_jobs, o_queue)
{:noreply, %{in: {i_running_jobs, i_queue}, out: {o_running_jobs, o_queue}}}
end end
def handle_cast(m, state) do def handle_cast(m, state) do
...@@ -106,9 +117,13 @@ def handle_cast(m, state) do ...@@ -106,9 +117,13 @@ def handle_cast(m, state) do
{:noreply, state} {:noreply, state}
end end
def handle_info({:DOWN, ref, :process, _pid, _reason}, {running_jobs, queue}) do def handle_info({:DOWN, ref, :process, _pid, _reason}, state) do
running_jobs = :sets.del_element(ref, running_jobs) %{in: {i_running_jobs, i_queue}, out: {o_running_jobs, o_queue}} = state
{running_jobs, queue} = maybe_start_job(running_jobs, queue) i_running_jobs = :sets.del_element(ref, i_running_jobs)
{:noreply, {running_jobs, queue}} o_running_jobs = :sets.del_element(ref, o_running_jobs)
{i_running_jobs, i_queue} = maybe_start_job(i_running_jobs, i_queue)
{o_running_jobs, o_queue} = maybe_start_job(o_running_jobs, o_queue)
{:noreply, %{in: {i_running_jobs, i_queue}, out: {o_running_jobs, o_queue}}}
end end
end end
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment