git.squeep.com Git - akkoma/blob - lib/pleroma/http/adapter_helper/gun.ex

   1 # Pleroma: A lightweight social networking server
   2 # Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
   3 # SPDX-License-Identifier: AGPL-3.0-only
   4
   5 defmodule Pleroma.HTTP.AdapterHelper.Gun do
   6   @behaviour Pleroma.HTTP.AdapterHelper
   7
   8   alias Pleroma.Config
   9   alias Pleroma.HTTP.AdapterHelper
  10
  11   require Logger
  12
  13   @defaults [
  14     connect_timeout: 5_000,
  15     domain_lookup_timeout: 5_000,
  16     tls_handshake_timeout: 5_000,
  17     retry: 1,
  18     retry_timeout: 1000,
  19     await_up_timeout: 5_000
  20   ]
  21
  22   @type pool() :: :federation | :upload | :media | :default
  23
  24   @spec options(keyword(), URI.t()) :: keyword()
  25   def options(incoming_opts \\ [], %URI{} = uri) do
  26     proxy =
  27       [:http, :proxy_url]
  28       |> Config.get()
  29       |> AdapterHelper.format_proxy()
  30
  31     config_opts = Config.get([:http, :adapter], [])
  32
  33     @defaults
  34     |> Keyword.merge(config_opts)
  35     |> add_scheme_opts(uri)
  36     |> AdapterHelper.maybe_add_proxy(proxy)
  37     |> Keyword.merge(incoming_opts)
  38     |> put_timeout()
  39   end
  40
  41   defp add_scheme_opts(opts, %{scheme: "http"}), do: opts
  42
  43   defp add_scheme_opts(opts, %{scheme: "https"}) do
  44     Keyword.put(opts, :certificates_verification, true)
  45   end
  46
  47   defp put_timeout(opts) do
  48     # this is the timeout to receive a message from Gun
  49     Keyword.put_new(opts, :timeout, pool_timeout(opts[:pool]))
  50   end
  51
  52   @spec pool_timeout(pool()) :: non_neg_integer()
  53   def pool_timeout(pool) do
  54     default = Config.get([:pools, :default, :timeout], 5_000)
  55
  56     Config.get([:pools, pool, :timeout], default)
  57   end
  58
  59   @prefix Pleroma.Gun.ConnectionPool
  60   def limiter_setup do
  61     wait = Config.get([:connections_pool, :connection_acquisition_wait])
  62     retries = Config.get([:connections_pool, :connection_acquisition_retries])
  63
  64     :pools
  65     |> Config.get([])
  66     |> Enum.each(fn {name, opts} ->
  67       max_running = Keyword.get(opts, :size, 50)
  68       max_waiting = Keyword.get(opts, :max_waiting, 10)
  69
  70       result =
  71         ConcurrentLimiter.new(:"#{@prefix}.#{name}", max_running, max_waiting,
  72           wait: wait,
  73           max_retries: retries
  74         )
  75
  76       case result do
  77         :ok -> :ok
  78         {:error, :existing} -> :ok
  79       end
  80     end)
  81
  82     :ok
  83   end
  84 end