1 # Pleroma: A lightweight social networking server
2 # Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
3 # SPDX-License-Identifier: AGPL-3.0-only
5 defmodule Pleroma.Gun.ConnectionPool do
8 alias Pleroma.Gun.ConnectionPool.WorkerSupervisor
12 {Registry, keys: :unique, name: @registry},
13 Pleroma.Gun.ConnectionPool.WorkerSupervisor
17 @spec get_conn(URI.t(), keyword()) :: {:ok, pid()} | {:error, term()}
18 def get_conn(uri, opts) do
19 key = "#{uri.scheme}:#{uri.host}:#{uri.port}"
21 case Registry.lookup(@registry, key) do
22 # The key has already been registered, but connection is not up yet
23 [{worker_pid, nil}] ->
24 get_gun_pid_from_worker(worker_pid, true)
26 [{worker_pid, {gun_pid, _used_by, _crf, _last_reference}}] ->
27 GenServer.call(worker_pid, :add_client)
31 # :gun.set_owner fails in :connected state for whatevever reason,
32 # so we open the connection in the process directly and send it's pid back
33 # We trust gun to handle timeouts by itself
34 case WorkerSupervisor.start_worker([key, uri, opts, self()]) do
36 get_gun_pid_from_worker(worker_pid, false)
38 {:error, {:already_started, worker_pid}} ->
39 get_gun_pid_from_worker(worker_pid, true)
47 defp get_gun_pid_from_worker(worker_pid, register) do
48 # GenServer.call will block the process for timeout length if
49 # the server crashes on startup (which will happen if gun fails to connect)
50 # so instead we use cast + monitor
52 ref = Process.monitor(worker_pid)
53 if register, do: GenServer.cast(worker_pid, {:add_client, self()})
57 Process.demonitor(ref)
60 {:DOWN, ^ref, :process, ^worker_pid, reason} ->
62 {:shutdown, {:error, _} = error} -> error
63 {:shutdown, error} -> {:error, error}
69 @spec release_conn(pid()) :: :ok
70 def release_conn(conn_pid) do
71 # :ets.fun2ms(fn {_, {worker_pid, {gun_pid, _, _, _}}} when gun_pid == conn_pid ->
74 Registry.select(@registry, [
75 {{:_, :"$1", {:"$2", :_, :_, :_}}, [{:==, :"$2", conn_pid}], [:"$1"]}
80 GenServer.call(worker_pid, :remove_client)