X-Git-Url: https://git.squeep.com/?a=blobdiff_plain;f=lib%2Fpleroma%2Fhttp%2Fadapter_helper%2Fgun.ex;h=b4ff8306c3adeabd505b28fa6440c3e91c0fd984;hb=6c59fe259d3d8eccc4102baaac64baf2aff301cb;hp=55c2b192aecf451010a29f7802ab555715543428;hpb=f39e1b9eff859c0795911212c59304f68fca92bc;p=akkoma diff --git a/lib/pleroma/http/adapter_helper/gun.ex b/lib/pleroma/http/adapter_helper/gun.ex index 55c2b192a..b4ff8306c 100644 --- a/lib/pleroma/http/adapter_helper/gun.ex +++ b/lib/pleroma/http/adapter_helper/gun.ex @@ -5,8 +5,8 @@ defmodule Pleroma.HTTP.AdapterHelper.Gun do @behaviour Pleroma.HTTP.AdapterHelper + alias Pleroma.Gun.ConnectionPool alias Pleroma.HTTP.AdapterHelper - alias Pleroma.Pool.Connections require Logger @@ -14,15 +14,15 @@ defmodule Pleroma.HTTP.AdapterHelper.Gun do connect_timeout: 5_000, domain_lookup_timeout: 5_000, tls_handshake_timeout: 5_000, - retry: 1, + retry: 0, retry_timeout: 1000, await_up_timeout: 5_000 ] @spec options(keyword(), URI.t()) :: keyword() - def options(connection_opts \\ [], %URI{} = uri) do - formatted_proxy = - Pleroma.Config.get([:http, :proxy_url], nil) + def options(incoming_opts \\ [], %URI{} = uri) do + proxy = + Pleroma.Config.get([:http, :proxy_url]) |> AdapterHelper.format_proxy() config_opts = Pleroma.Config.get([:http, :adapter], []) @@ -30,57 +30,49 @@ defmodule Pleroma.HTTP.AdapterHelper.Gun do @defaults |> Keyword.merge(config_opts) |> add_scheme_opts(uri) - |> AdapterHelper.maybe_add_proxy(formatted_proxy) - |> maybe_get_conn(uri, connection_opts) + |> AdapterHelper.maybe_add_proxy(proxy) + |> Keyword.merge(incoming_opts) end - @spec after_request(keyword()) :: :ok - def after_request(opts) do - if opts[:conn] && opts[:body_as] != :chunks do - Connections.checkout(opts[:conn], self(), :gun_connections) - end - - :ok - end - - defp add_scheme_opts(opts, %URI{scheme: "http"}), do: opts + defp add_scheme_opts(opts, %{scheme: "http"}), do: opts - defp add_scheme_opts(opts, %URI{scheme: "https"}) do + defp add_scheme_opts(opts, %{scheme: "https"}) do opts |> Keyword.put(:certificates_verification, true) - |> Keyword.put(:transport, :tls) - |> Keyword.put(:tls_opts, log_level: :warning) end - defp maybe_get_conn(adapter_opts, uri, connection_opts) do - {receive_conn?, opts} = - adapter_opts - |> Keyword.merge(connection_opts) - |> Keyword.pop(:receive_conn, true) - - if Connections.alive?(:gun_connections) and receive_conn? do - try_to_get_conn(uri, opts) - else - opts + @spec get_conn(URI.t(), keyword()) :: {:ok, keyword()} | {:error, atom()} + def get_conn(uri, opts) do + case ConnectionPool.get_conn(uri, opts) do + {:ok, conn_pid} -> {:ok, Keyword.merge(opts, conn: conn_pid, close_conn: false)} + err -> err end end - defp try_to_get_conn(uri, opts) do - case Connections.checkin(uri, :gun_connections) do - nil -> - Logger.debug( - "Gun connections pool checkin was not successful. Trying to open conn for next request." + @prefix Pleroma.Gun.ConnectionPool + def limiter_setup do + wait = Pleroma.Config.get([:connections_pool, :connection_acquisition_wait]) + retries = Pleroma.Config.get([:connections_pool, :connection_acquisition_retries]) + + :pools + |> Pleroma.Config.get([]) + |> Enum.each(fn {name, opts} -> + max_running = Keyword.get(opts, :size, 50) + max_waiting = Keyword.get(opts, :max_waiting, 10) + + result = + ConcurrentLimiter.new(:"#{@prefix}.#{name}", max_running, max_waiting, + wait: wait, + max_retries: retries ) - Task.start(fn -> Pleroma.Gun.Conn.open(uri, :gun_connections, opts) end) - opts - - conn when is_pid(conn) -> - Logger.debug("received conn #{inspect(conn)} #{Connections.compose_uri_log(uri)}") + case result do + :ok -> :ok + {:error, :existing} -> :ok + e -> raise e + end + end) - opts - |> Keyword.put(:conn, conn) - |> Keyword.put(:close_conn, false) - end + :ok end end