X-Git-Url: http://git.squeep.com/?a=blobdiff_plain;f=lib%2Fpleroma%2Fgun%2Fconnection_pool.ex;h=8b41a668c0268c504ec55ddc05bbb7545aedcffe;hb=10d46235cfae03901c297473043a5700a8c5e4f0;hp=545bfaf7fea486564d9d035914525c7575463e50;hpb=ec9d0d146b4ec6752f8f2896ace9bb5585469773;p=akkoma diff --git a/lib/pleroma/gun/connection_pool.ex b/lib/pleroma/gun/connection_pool.ex index 545bfaf7f..8b41a668c 100644 --- a/lib/pleroma/gun/connection_pool.ex +++ b/lib/pleroma/gun/connection_pool.ex @@ -15,8 +15,8 @@ defmodule Pleroma.Gun.ConnectionPool do case Registry.lookup(@registry, key) do # The key has already been registered, but connection is not up yet - [{worker_pid, {nil, _used_by, _crf, _last_reference}}] -> - get_gun_pid_from_worker(worker_pid) + [{worker_pid, nil}] -> + get_gun_pid_from_worker(worker_pid, true) [{worker_pid, {gun_pid, _used_by, _crf, _last_reference}}] -> GenServer.cast(worker_pid, {:add_client, self(), false}) @@ -26,14 +26,12 @@ defmodule Pleroma.Gun.ConnectionPool do # :gun.set_owner fails in :connected state for whatevever reason, # so we open the connection in the process directly and send it's pid back # We trust gun to handle timeouts by itself - case WorkerSupervisor.start_worker([uri, key, opts, self()]) do - {:ok, _worker_pid} -> - receive do - {:conn_pid, pid} -> {:ok, pid} - end + case WorkerSupervisor.start_worker([key, uri, opts, self()]) do + {:ok, worker_pid} -> + get_gun_pid_from_worker(worker_pid, false) - {:error, {:error, {:already_registered, worker_pid}}} -> - get_gun_pid_from_worker(worker_pid) + {:error, {:already_started, worker_pid}} -> + get_gun_pid_from_worker(worker_pid, true) err -> err @@ -41,21 +39,30 @@ defmodule Pleroma.Gun.ConnectionPool do end end - defp get_gun_pid_from_worker(worker_pid) do + defp get_gun_pid_from_worker(worker_pid, register) do # GenServer.call will block the process for timeout length if # the server crashes on startup (which will happen if gun fails to connect) # so instead we use cast + monitor ref = Process.monitor(worker_pid) - GenServer.cast(worker_pid, {:add_client, self(), true}) + if register, do: GenServer.cast(worker_pid, {:add_client, self(), true}) receive do - {:conn_pid, pid} -> {:ok, pid} - {:DOWN, ^ref, :process, ^worker_pid, reason} -> reason + {:conn_pid, pid} -> + Process.demonitor(ref) + {:ok, pid} + + {:DOWN, ^ref, :process, ^worker_pid, reason} -> + case reason do + {:shutdown, error} -> error + _ -> {:error, reason} + end end end def release_conn(conn_pid) do + # :ets.fun2ms(fn {_, {worker_pid, {gun_pid, _, _, _}}} when gun_pid == conn_pid -> + # worker_pid end) query_result = Registry.select(@registry, [ {{:_, :"$1", {:"$2", :_, :_, :_}}, [{:==, :"$2", conn_pid}], [:"$1"]}