Elixir HTTP 客户端与连接池：Mint、Finch 与 Req 实战

{:ok, conn} = Mint.HTTP.connect(:https, "api.example.com", 443,
  protocols: [:http2, :http1],
  transport_opts: [verify: :verify_peer, cacerts: :public_key.cacerts_get()]
)

IO.inspect(Mint.HTTP.protocol(conn))   # :http2 或 :http1

{:ok, conn, request_ref} =
  Mint.HTTP.request(conn, "GET", "/v1/users", [{"accept", "application/json"}], nil)

receive_loop(conn, request_ref, %{}).

defp receive_loop(conn, ref, acc) do
  receive do
    message ->
      {:ok, conn, responses} = Mint.HTTP.stream(conn, message)

      case Enum.reduce(responses, acc, &handle_response(&1, ref, &2)) do
        {:done, body} -> {conn, body}
        acc -> receive_loop(conn, ref, acc)
      end
  after
    5_000 -> {:error, :timeout}
  end
end

defp handle_response({:status, ^ref, status}, _ref, acc), do: Map.put(acc, :status, status)
defp handle_response({:headers, ^ref, h}, _ref, acc), do: Map.put(acc, :headers, h)
defp handle_response({:data, ^ref, chunk}, _ref, acc), do: Map.update(acc, :body, chunk, &(&1 <> chunk))
defp handle_response({:done, ^ref}, _ref, acc), do: {:done, Map.get(acc, :body, "")}
defp handle_response(_, _, acc), do: acc

children = [
  {Finch,
   name: MyFinch,
   pools: %{
     default: [size: 50, count: 1],
     "https://api.example.com" => [size: 100, count: 4],
     "https://slow-partner.com" => [size: 10, count: 1]
   }}
]

pools: %{
  default: [size: 10, count: 1],
  # 内部服务：低延迟、高容量
  "http://internal-svc.default.svc.cluster.local" => [size: 100, count: 2],
  # 外部支付网关：严格限流，池要小
  "https://api.payment.com" => [size: 5, count: 1,
                                conn_opts: [transport_opts: [timeout: 3_000]]]
}

# 普通请求
request = Finch.build(:get, "https://api.example.com/v1/users", [{"accept", "application/json"}])
{:ok, %Finch.Response{status: 200, body: body}} = Finch.request(request, MyFinch, receive_timeout: 5_000)

# 流式处理：避免把大响应体整个读进内存
Finch.stream(request, MyFinch, [], fn
  {:status, status}, acc -> [{:status, status} | acc]
  {:headers, headers}, acc -> [{:headers, headers} | acc]
  {:data, chunk}, acc -> [{:data, byte_size(chunk)} | acc]
end)

# GET + 自动 JSON 解码
{:ok, %Req.Response{status: 200, body: %{"users" => users}}} =
  Req.get("https://api.example.com/v1/users")

# POST JSON
Req.post!("https://api.example.com/v1/users",
  json: %{name: "Alice", email: "a@b.c"},
  headers: [{"authorization", "Bearer #{token}"}],
  receive_timeout: 5_000)

Req.get!(url,
  retry: :transient,        # 只重试「瞬时错误」：5xx、超时、连接错误
  max_retries: 3,
  retry_delay: fn attempt -> trunc(:math.pow(2, attempt) * 100) + :rand.uniform(50) end,
  retry_log_level: :warning
)

Req.new(url: "https://api.example.com/v1/users")
|> Req.Request.append_request_steps(sign: &sign_request/1)
|> Req.Request.append_response_steps(record: &record_metrics/1)
|> Req.get!()

defp sign_request(request) do
  signature = :crypto.mac(:hmac, :sha256, secret(), request.body || "")
  Req.Request.put_header(request, "x-signature", Base.encode16(signature, case: :lower))
end

Req.get!(url, plugins: [
  {Req.Finch, finch: MyFinch},              # 指定 Finch 实例
  {Req.Steps.retry, retry: :transient},     # 重试
  &Req.Steps.put_base_url/1                 # 相对路径
])

# Basic Auth
Req.get!(url, auth: {:basic, "user:pass"})

# Bearer Token
Req.get!(url, auth: {:bearer, token})

# AWS SigV4（通过插件）
Req.get!(url, aws_sigv4: [service: :s3, region: "ap-northeast-1"])

Req.get!(url,
  connect_options: [timeout: 3_000],
  pool_timeout: 5_000,
  receive_timeout: 10_000)

defmodule MyApp.CircuitBreaker do
  use GenServer

  @threshold 5           # 连续失败阈值
  @reset_after 30_000    # 熔断后多久尝试恢复

  def call(fun) do
    case :ets.lookup(:breaker, :payment) do
      [{:payment, :open, opened_at}] when opened_at + @reset_after > now() ->
        {:error, :circuit_open}
      _ ->
        case fun.() do
          {:ok, _} = ok -> record_success(); ok
          {:error, _} = err -> record_failure(); err
        end
    end
  end
end

pools: %{
  "https://api.payment.com" => [size: 5, count: 1],     # 最多 5 个并发
  "https://api.search.com"  => [size: 50, count: 2]     # 最多 100 个并发
}

# 下载大文件：边收边写盘，内存恒定
File.open!("big.zip", [:write])
|> then(fn file ->
  Finch.stream(Finch.build(:get, url), MyFinch, file, fn
    {:data, chunk}, file -> IO.binwrite(file, chunk); file
    _, file -> file
  end)
end)

# SSE：逐事件消费
Finch.stream(request, MyFinch, "", fn
  {:data, chunk}, acc ->
    case String.split(acc <> chunk, "\n\n") do
      [rest] -> rest
      parts -> handle_events(Enum.drop(parts, -1)); List.last(parts)
    end
  _, acc -> acc
end)

:telemetry.attach_many("http-metrics",
  [[:finch, :request, :stop], [:finch, :request, :exception]],
  fn event, measurements, metadata, _cfg ->
    MyApp.Metrics.record(event, measurements, metadata.request.host)
  end, nil)