zf

http1.ex (32953B)

---

 defmodule Mint.HTTP1 do
   @moduledoc """
   Processless HTTP client with support for HTTP/1 and HTTP/1.1.
 
   This module provides a data structure that represents an HTTP/1 or HTTP/1.1 connection to
   a given server. The connection is represented as an opaque struct `%Mint.HTTP1{}`.
   The connection is a data structure and is not backed by a process, and all the
   connection handling happens in the process that creates the struct.
 
   This module and data structure work exactly like the ones described in the `Mint`
   module, with the exception that `Mint.HTTP1` specifically deals with HTTP/1 and HTTP/1.1 while
   `Mint` deals seamlessly with HTTP/1, HTTP/1.1, and HTTP/2. For more information on
   how to use the data structure and client architecture, see `Mint`.
   """
 
   import Mint.Core.Util
 
   alias Mint.Core.Util
   alias Mint.HTTP1.{Parse, Request, Response}
   alias Mint.{HTTPError, TransportError, Types}
 
   require Logger
 
   @behaviour Mint.Core.Conn
 
   @opaque t() :: %__MODULE__{}
 
   @user_agent "mint/" <> Mix.Project.config()[:version]
 
   @typedoc """
   An HTTP/1-specific error reason.
 
   The values can be:
 
     * `:closed` - when you try to make a request or stream a body chunk but the connection
       is closed.
 
     * `:request_body_is_streaming` - when you call `request/5` to send a new
       request but another request is already streaming.
 
     * `{:unexpected_data, data}` - when unexpected data is received from the server.
 
     * `:invalid_status_line` - when the HTTP/1 status line is invalid.
 
     * `{:invalid_request_target, target}` - when the request target is invalid.
 
     * `:invalid_header` - when headers can't be parsed correctly.
 
     * `{:invalid_header_name, name}` - when a header name is invalid.
 
     * `{:invalid_header_value, name, value}` - when a header value is invalid. `name`
       is the name of the header and `value` is the invalid value.
 
     * `:invalid_chunk_size` - when the chunk size is invalid.
 
     * `:missing_crlf_after_chunk` - when the CRLF after a chunk is missing.
 
     * `:invalid_trailer_header` - when trailer headers can't be parsed.
 
     * `:more_than_one_content_length_header` - when more than one `content-length`
       headers are in the response.
 
     * `:transfer_encoding_and_content_length` - when both the `content-length` as well
       as the `transfer-encoding` headers are in the response.
 
     * `{:invalid_content_length_header, value}` - when the value of the `content-length`
       header is invalid, that is, is not an non-negative integer.
 
     * `:empty_token_list` - when a header that is supposed to contain a list of tokens
       (such as the `connection` header) doesn't contain any.
 
     * `{:invalid_token_list, string}` - when a header that is supposed to contain a list
       of tokens (such as the `connection` header) contains a malformed list of tokens.
 
     * `:trailing_headers_but_not_chunked_encoding` - when you try to send trailing
       headers through `stream_request_body/3` but the transfer encoding of the request
       was not `chunked`.
 
   """
   @type error_reason() :: term()
 
   defstruct [
     :host,
     :port,
     :request,
     :streaming_request,
     :socket,
     :transport,
     :mode,
     :scheme_as_string,
     requests: :queue.new(),
     state: :closed,
     buffer: "",
     proxy_headers: [],
     private: %{}
   ]
 
   @doc """
   Same as `Mint.HTTP.connect/4`, but forces an HTTP/1 or HTTP/1.1 connection.
 
   This function doesn't support proxying.
   """
   @spec connect(Types.scheme(), Types.address(), :inet.port_number(), keyword()) ::
           {:ok, t()} | {:error, Types.error()}
   def connect(scheme, address, port, opts \\ []) do
     # TODO: Also ALPN negotiate HTTP1?
 
     hostname = Mint.Core.Util.hostname(opts, address)
     transport = scheme_to_transport(scheme)
 
     transport_opts =
       Keyword.get(opts, :transport_opts, [])
       |> Keyword.put(:hostname, hostname)
 
     with {:ok, socket} <- transport.connect(address, port, transport_opts) do
       initiate(scheme, socket, hostname, port, opts)
     end
   end
 
   @doc false
   @spec upgrade(
           Types.scheme(),
           Types.socket(),
           Types.scheme(),
           String.t(),
           :inet.port_number(),
           keyword()
         ) :: {:ok, t()} | {:error, Types.error()}
   def upgrade(old_scheme, socket, new_scheme, hostname, port, opts) do
     # TODO: Also ALPN negotiate HTTP1?
 
     transport = scheme_to_transport(new_scheme)
 
     transport_opts =
       Keyword.get(opts, :transport_opts, [])
       |> Keyword.put(:hostname, hostname)
 
     with {:ok, socket} <- transport.upgrade(socket, old_scheme, hostname, port, transport_opts) do
       initiate(new_scheme, socket, hostname, port, opts)
     end
   end
 
   @doc false
   @impl true
   @spec initiate(
           Types.scheme(),
           Types.socket(),
           String.t(),
           :inet.port_number(),
           keyword()
         ) :: {:ok, t()} | {:error, Types.error()}
   def initiate(scheme, socket, hostname, port, opts) do
     transport = scheme_to_transport(scheme)
     mode = Keyword.get(opts, :mode, :active)
 
     unless mode in [:active, :passive] do
       raise ArgumentError,
             "the :mode option must be either :active or :passive, got: #{inspect(mode)}"
     end
 
     with :ok <- inet_opts(transport, socket),
          :ok <- if(mode == :active, do: transport.setopts(socket, active: :once), else: :ok) do
       conn = %__MODULE__{
         transport: transport,
         socket: socket,
         mode: mode,
         host: hostname,
         port: port,
         scheme_as_string: Atom.to_string(scheme),
         state: :open
       }
 
       {:ok, conn}
     else
       {:error, reason} ->
         :ok = transport.close(socket)
         {:error, reason}
     end
   end
 
   @doc """
   See `Mint.HTTP.close/1`.
   """
   @impl true
   @spec close(t()) :: {:ok, t()}
   def close(conn)
 
   def close(%__MODULE__{state: :open} = conn) do
     conn = internal_close(conn)
     {:ok, conn}
   end
 
   def close(%__MODULE__{state: :closed} = conn) do
     {:ok, conn}
   end
 
   @doc """
   See `Mint.HTTP.open?/1`.
   """
   @impl true
   @spec open?(t(), :read | :write | :read_write) :: boolean()
   def open?(conn, type \\ :read_write)
 
   def open?(%__MODULE__{state: state}, type) when type in [:read, :write, :read_write] do
     state == :open
   end
 
   @doc """
   See `Mint.HTTP.request/5`.
 
   In HTTP/1 and HTTP/1.1, you can't open a new request if you're streaming the body of
   another request. If you try, an error will be returned.
   """
   @impl true
   @spec request(
           t(),
           method :: String.t(),
           path :: String.t(),
           Types.headers(),
           body :: iodata() | nil | :stream
         ) ::
           {:ok, t(), Types.request_ref()}
           | {:error, t(), Types.error()}
   def request(conn, method, path, headers, body)
 
   def request(%__MODULE__{state: :closed} = conn, _method, _path, _headers, _body) do
     {:error, conn, wrap_error(:closed)}
   end
 
   def request(%__MODULE__{streaming_request: %{}} = conn, _method, _path, _headers, _body) do
     {:error, conn, wrap_error(:request_body_is_streaming)}
   end
 
   def request(%__MODULE__{} = conn, method, path, headers, body) do
     %__MODULE__{transport: transport, socket: socket} = conn
 
     headers =
       headers
       |> lower_header_keys()
       |> add_default_headers(conn)
 
     with {:ok, headers, encoding} <- add_content_length_or_transfer_encoding(headers, body),
          {:ok, iodata} <- Request.encode(method, path, headers, body),
          :ok <- transport.send(socket, iodata) do
       request_ref = make_ref()
       request = new_request(request_ref, method, body, encoding)
 
       case request.state do
         {:stream_request, _} ->
           conn = %__MODULE__{conn | streaming_request: request}
           {:ok, conn, request_ref}
 
         _ ->
           conn = enqueue_request(conn, request)
           {:ok, conn, request_ref}
       end
     else
       {:error, %TransportError{reason: :closed} = error} ->
         {:error, %{conn | state: :closed}, error}
 
       {:error, %error_module{} = error} when error_module in [HTTPError, TransportError] ->
         {:error, conn, error}
 
       {:error, reason} ->
         {:error, conn, wrap_error(reason)}
     end
   end
 
   @doc """
   See `Mint.HTTP.stream_request_body/3`.
 
   In HTTP/1, sending an empty chunk is a no-op.
 
   ## Transfer encoding and content length
 
   When streaming the request body, Mint cannot send a precalculated `content-length`
   request header because it doesn't know the body that you'll stream. However, Mint
   will transparently handle the presence of a `content-length` header using this logic:
 
     * if you specifically set a `content-length` header, then transfer encoding and
       making sure the content length is correct for what you'll stream is up to you.
 
     * if you specifically set the transfer encoding (`transfer-encoding` header)
       to `chunked`, then it's up to you to
       [properly encode chunks](https://en.wikipedia.org/wiki/Chunked_transfer_encoding).
 
     * if you don't set the transfer encoding to `chunked` and don't provide a
       `content-length` header, Mint will do implicit `chunked` transfer encoding
       (setting the `transfer-encoding` header appropriately) and will take care
       of properly encoding the chunks.
 
   """
   @impl true
   @spec stream_request_body(
           t(),
           Types.request_ref(),
           iodata() | :eof | {:eof, trailing_headers :: Types.headers()}
         ) ::
           {:ok, t()} | {:error, t(), Types.error()}
   def stream_request_body(
         %__MODULE__{streaming_request: %{state: {:stream_request, :identity}, ref: ref}} = conn,
         ref,
         :eof
       ) do
     request = %{conn.streaming_request | state: :status}
     conn = enqueue_request(%__MODULE__{conn | streaming_request: nil}, request)
     {:ok, conn}
   end
 
   def stream_request_body(
         %__MODULE__{streaming_request: %{state: {:stream_request, :identity}, ref: ref}} = conn,
         ref,
         {:eof, _trailing_headers}
       ) do
     {:error, conn, wrap_error(:trailing_headers_but_not_chunked_encoding)}
   end
 
   def stream_request_body(
         %__MODULE__{streaming_request: %{state: {:stream_request, :identity}, ref: ref}} = conn,
         ref,
         body
       ) do
     case conn.transport.send(conn.socket, body) do
       :ok ->
         {:ok, conn}
 
       {:error, %TransportError{reason: :closed} = error} ->
         {:error, %{conn | state: :closed}, error}
 
       {:error, error} ->
         {:error, conn, error}
     end
   end
 
   def stream_request_body(
         %__MODULE__{streaming_request: %{state: {:stream_request, :chunked}, ref: ref}} = conn,
         ref,
         chunk
       ) do
     with {:ok, chunk} <- validate_chunk(chunk),
          :ok <- conn.transport.send(conn.socket, Request.encode_chunk(chunk)) do
       case chunk do
         :eof ->
           request = %{conn.streaming_request | state: :status}
           conn = enqueue_request(%__MODULE__{conn | streaming_request: nil}, request)
           {:ok, conn}
 
         {:eof, _trailing_headers} ->
           request = %{conn.streaming_request | state: :status}
           conn = enqueue_request(%__MODULE__{conn | streaming_request: nil}, request)
           {:ok, conn}
 
         _other ->
           {:ok, conn}
       end
     else
       :empty_chunk ->
         {:ok, conn}
 
       {:error, %TransportError{reason: :closed} = error} ->
         {:error, %{conn | state: :closed}, error}
 
       {:error, error} ->
         {:error, conn, error}
     end
   end
 
   defp validate_chunk({:eof, trailing_headers}) do
     headers = lower_header_keys(trailing_headers)
 
     if unallowed_header = find_unallowed_trailing_header(headers) do
       {:error, wrap_error({:unallowed_trailing_header, unallowed_header})}
     else
       {:ok, {:eof, headers}}
     end
   end
 
   defp validate_chunk(:eof) do
     {:ok, :eof}
   end
 
   defp validate_chunk(chunk) do
     if IO.iodata_length(chunk) == 0 do
       :empty_chunk
     else
       {:ok, chunk}
     end
   end
 
   @doc """
   See `Mint.HTTP.stream/2`.
   """
   @impl true
   @spec stream(t(), term()) ::
           {:ok, t(), [Types.response()]}
           | {:error, t(), Types.error(), [Types.response()]}
           | :unknown
   def stream(conn, message)
 
   def stream(%__MODULE__{transport: transport, socket: socket} = conn, {tag, socket, data})
       when tag in [:tcp, :ssl] do
     case handle_data(conn, data) do
       {:ok, %{mode: mode, state: state} = conn, responses}
       when mode == :active and state != :closed ->
         case transport.setopts(socket, active: :once) do
           :ok -> {:ok, conn, responses}
           {:error, reason} -> {:error, put_in(conn.state, :closed), reason, responses}
         end
 
       other ->
         other
     end
   end
 
   def stream(%__MODULE__{socket: socket} = conn, {tag, socket})
       when tag in [:tcp_closed, :ssl_closed] do
     handle_close(conn)
   end
 
   def stream(%__MODULE__{socket: socket} = conn, {tag, socket, reason})
       when tag in [:tcp_error, :ssl_error] do
     handle_error(conn, conn.transport.wrap_error(reason))
   end
 
   def stream(%__MODULE__{}, _message) do
     :unknown
   end
 
   defp handle_data(%__MODULE__{request: nil} = conn, data) do
     conn = internal_close(conn)
     {:error, conn, wrap_error({:unexpected_data, data}), []}
   end
 
   defp handle_data(%__MODULE__{request: request} = conn, data) do
     data = maybe_concat(conn.buffer, data)
 
     case decode(request.state, conn, data, []) do
       {:ok, conn, responses} ->
         {:ok, conn, Enum.reverse(responses)}
 
       {:error, conn, reason, responses} ->
         conn = put_in(conn.state, :closed)
         {:error, conn, reason, responses}
     end
   end
 
   defp handle_close(%__MODULE__{request: request} = conn) do
     conn = put_in(conn.state, :closed)
     conn = request_done(conn)
 
     if request && request.body == :until_closed do
       conn = put_in(conn.state, :closed)
       {:ok, conn, [{:done, request.ref}]}
     else
       {:error, conn, conn.transport.wrap_error(:closed), []}
     end
   end
 
   defp handle_error(conn, error) do
     conn = put_in(conn.state, :closed)
     {:error, conn, error, []}
   end
 
   @doc """
   See `Mint.HTTP.recv/3`.
   """
   @impl true
   @spec recv(t(), non_neg_integer(), timeout()) ::
           {:ok, t(), [Types.response()]}
           | {:error, t(), Types.error(), [Types.response()]}
   def recv(conn, byte_count, timeout)
 
   def recv(%__MODULE__{mode: :passive} = conn, byte_count, timeout) do
     case conn.transport.recv(conn.socket, byte_count, timeout) do
       {:ok, data} -> handle_data(conn, data)
       {:error, %Mint.TransportError{reason: :closed}} -> handle_close(conn)
       {:error, error} -> handle_error(conn, error)
     end
   end
 
   def recv(_conn, _byte_count, _timeout) do
     raise ArgumentError,
           "can't use recv/3 to synchronously receive data when the mode is :active. " <>
             "Use Mint.HTTP.set_mode/2 to set the connection to passive mode"
   end
 
   @doc """
   See `Mint.HTTP.set_mode/2`.
   """
   @impl true
   @spec set_mode(t(), :active | :passive) :: {:ok, t()} | {:error, Types.error()}
   def set_mode(%__MODULE__{} = conn, mode) when mode in [:active, :passive] do
     active =
       case mode do
         :active -> :once
         :passive -> false
       end
 
     with :ok <- conn.transport.setopts(conn.socket, active: active) do
       {:ok, put_in(conn.mode, mode)}
     end
   end
 
   @doc """
   See `Mint.HTTP.controlling_process/2`.
   """
   @impl true
   @spec controlling_process(t(), pid()) :: {:ok, t()} | {:error, Types.error()}
   def controlling_process(%__MODULE__{} = conn, new_pid) when is_pid(new_pid) do
     with :ok <- conn.transport.controlling_process(conn.socket, new_pid) do
       {:ok, conn}
     end
   end
 
   @doc """
   See `Mint.HTTP.open_request_count/1`.
 
   In HTTP/1, the number of open requests is the number of pipelined requests.
   """
   @impl true
   @spec open_request_count(t()) :: non_neg_integer()
   def open_request_count(%__MODULE__{} = conn) do
     case conn do
       %{request: nil, streaming_request: nil} -> 0
       %{request: nil} -> 1
       %{streaming_request: nil} -> 1 + :queue.len(conn.requests)
       _ -> 2 + :queue.len(conn.requests)
     end
   end
 
   @doc """
   See `Mint.HTTP.put_private/3`.
   """
   @impl true
   @spec put_private(t(), atom(), term()) :: t()
   def put_private(%__MODULE__{private: private} = conn, key, value) when is_atom(key) do
     %{conn | private: Map.put(private, key, value)}
   end
 
   @doc """
   See `Mint.HTTP.get_private/3`.
   """
   @impl true
   @spec get_private(t(), atom(), term()) :: term()
   def get_private(%__MODULE__{private: private} = _conn, key, default \\ nil) when is_atom(key) do
     Map.get(private, key, default)
   end
 
   @doc """
   See `Mint.HTTP.delete_private/2`.
   """
   @impl true
   @spec delete_private(t(), atom()) :: t()
   def delete_private(%__MODULE__{private: private} = conn, key) when is_atom(key) do
     %{conn | private: Map.delete(private, key)}
   end
 
   @doc """
   See `Mint.HTTP.get_socket/1`.
   """
   @impl true
   @spec get_socket(t()) :: Mint.Types.socket()
   def get_socket(%__MODULE__{socket: socket} = _conn) do
     socket
   end
 
   @doc """
   See `Mint.HTTP.get_proxy_headers/1`.
   """
   if Version.compare(System.version(), "1.7.0") in [:eq, :gt] do
     @doc since: "1.4.0"
   end
 
   @impl true
   @spec get_proxy_headers(t()) :: Mint.Types.headers()
   def get_proxy_headers(%__MODULE__{proxy_headers: proxy_headers}), do: proxy_headers
 
   ## Helpers
 
   defp decode(:status, %{request: request} = conn, data, responses) do
     case Response.decode_status_line(data) do
       {:ok, {version, status, _reason}, rest} ->
         request = %{request | version: version, status: status, state: :headers}
         conn = %{conn | request: request}
         responses = [{:status, request.ref, status} | responses]
         decode(:headers, conn, rest, responses)
 
       :more ->
         conn = put_in(conn.buffer, data)
         {:ok, conn, responses}
 
       :error ->
         {:error, conn, wrap_error(:invalid_status_line), responses}
     end
   end
 
   defp decode(:headers, %{request: request} = conn, data, responses) do
     decode_headers(conn, request, data, responses, request.headers_buffer)
   end
 
   defp decode(:body, conn, data, responses) do
     case message_body(conn.request) do
       {:ok, body} ->
         conn = put_in(conn.request.body, body)
         decode_body(body, conn, data, conn.request.ref, responses)
 
       {:error, reason} ->
         {:error, conn, wrap_error(reason), responses}
     end
   end
 
   defp decode_headers(conn, request, data, responses, headers) do
     case Response.decode_header(data) do
       {:ok, {name, value}, rest} ->
         headers = [{name, value} | headers]
 
         case store_header(request, name, value) do
           {:ok, request} -> decode_headers(conn, request, rest, responses, headers)
           {:error, reason} -> {:error, conn, wrap_error(reason), responses}
         end
 
       {:ok, :eof, rest} ->
         responses = [{:headers, request.ref, Enum.reverse(headers)} | responses]
         request = %{request | state: :body, headers_buffer: []}
         conn = %{conn | buffer: "", request: request}
         decode(:body, conn, rest, responses)
 
       :more ->
         request = %{request | headers_buffer: headers}
         conn = %{conn | buffer: data, request: request}
         {:ok, conn, responses}
 
       :error ->
         {:error, conn, wrap_error(:invalid_header), responses}
     end
   end
 
   defp decode_body(:none, conn, data, request_ref, responses) do
     conn = put_in(conn.buffer, data)
     conn = request_done(conn)
     responses = [{:done, request_ref} | responses]
     {:ok, conn, responses}
   end
 
   defp decode_body(:single, conn, data, request_ref, responses) do
     {conn, responses} = add_body(conn, data, responses)
     conn = request_done(conn)
     responses = [{:done, request_ref} | responses]
     {:ok, conn, responses}
   end
 
   defp decode_body(:until_closed, conn, data, _request_ref, responses) do
     {conn, responses} = add_body(conn, data, responses)
     {:ok, conn, responses}
   end
 
   defp decode_body({:content_length, length}, conn, data, request_ref, responses) do
     cond do
       length > byte_size(data) ->
         conn = put_in(conn.request.body, {:content_length, length - byte_size(data)})
         {conn, responses} = add_body(conn, data, responses)
         {:ok, conn, responses}
 
       length <= byte_size(data) ->
         <<body::binary-size(length), rest::binary>> = data
         {conn, responses} = add_body(conn, body, responses)
         conn = request_done(conn)
         responses = [{:done, request_ref} | responses]
         next_request(conn, rest, responses)
     end
   end
 
   defp decode_body({:chunked, nil}, conn, "", _request_ref, responses) do
     conn = put_in(conn.buffer, "")
     conn = put_in(conn.request.body, {:chunked, nil})
     {:ok, conn, responses}
   end
 
   defp decode_body({:chunked, nil}, conn, data, request_ref, responses) do
     case Integer.parse(data, 16) do
       {_size, ""} ->
         conn = put_in(conn.buffer, data)
         conn = put_in(conn.request.body, {:chunked, nil})
         {:ok, conn, responses}
 
       {0, rest} ->
         # Manually collapse the body buffer since we're done with the body
         {conn, responses} = collapse_body_buffer(conn, responses)
         decode_body({:chunked, :metadata, :trailer}, conn, rest, request_ref, responses)
 
       {size, rest} when size > 0 ->
         decode_body({:chunked, :metadata, size}, conn, rest, request_ref, responses)
 
       _other ->
         {:error, conn, wrap_error(:invalid_chunk_size), responses}
     end
   end
 
   defp decode_body({:chunked, :metadata, size}, conn, data, request_ref, responses) do
     case Parse.ignore_until_crlf(data) do
       {:ok, rest} ->
         decode_body({:chunked, size}, conn, rest, request_ref, responses)
 
       :more ->
         conn = put_in(conn.buffer, data)
         conn = put_in(conn.request.body, {:chunked, :metadata, size})
         {:ok, conn, responses}
     end
   end
 
   defp decode_body({:chunked, :trailer}, conn, data, _request_ref, responses) do
     decode_trailer_headers(conn, data, responses, conn.request.headers_buffer)
   end
 
   defp decode_body({:chunked, :crlf}, conn, data, request_ref, responses) do
     case data do
       <<"\r\n", rest::binary>> ->
         conn = put_in(conn.request.body, {:chunked, nil})
         decode_body({:chunked, nil}, conn, rest, request_ref, responses)
 
       _other when byte_size(data) < 2 ->
         conn = put_in(conn.buffer, data)
         {:ok, conn, responses}
 
       _other ->
         {:error, conn, wrap_error(:missing_crlf_after_chunk), responses}
     end
   end
 
   defp decode_body({:chunked, length}, conn, data, request_ref, responses) do
     cond do
       length > byte_size(data) ->
         conn = put_in(conn.buffer, "")
         conn = put_in(conn.request.body, {:chunked, length - byte_size(data)})
         conn = add_body_to_buffer(conn, data)
         {:ok, conn, responses}
 
       length <= byte_size(data) ->
         <<body::binary-size(length), rest::binary>> = data
         {conn, responses} = add_body(conn, body, responses)
         conn = put_in(conn.request.body, {:chunked, :crlf})
         decode_body({:chunked, :crlf}, conn, rest, request_ref, responses)
     end
   end
 
   defp decode_trailer_headers(conn, data, responses, headers) do
     case Response.decode_header(data) do
       {:ok, {name, value}, rest} ->
         headers = [{name, value} | headers]
         decode_trailer_headers(conn, rest, responses, headers)
 
       {:ok, :eof, rest} ->
         headers = Util.remove_unallowed_trailing_headers(headers)
 
         responses = [
           {:done, conn.request.ref}
           | add_trailing_headers(headers, conn.request.ref, responses)
         ]
 
         conn = request_done(conn)
         next_request(conn, rest, responses)
 
       :more ->
         request = %{conn.request | body: {:chunked, :trailer}, headers_buffer: headers}
         conn = %{conn | buffer: data, request: request}
         {:ok, conn, responses}
 
       :error ->
         {:error, conn, wrap_error(:invalid_trailer_header), responses}
     end
   end
 
   defp next_request(%{request: nil} = conn, data, responses) do
     # TODO: Figure out if we should keep buffering even though there are no
     # requests in flight
     {:ok, %{conn | buffer: data}, responses}
   end
 
   defp next_request(conn, data, responses) do
     decode(:status, %{conn | state: :status}, data, responses)
   end
 
   defp add_trailing_headers([], _request_ref, responses), do: responses
 
   defp add_trailing_headers(headers, request_ref, responses),
     do: [{:headers, request_ref, Enum.reverse(headers)} | responses]
 
   defp add_body(conn, data, responses) do
     conn = add_body_to_buffer(conn, data)
     collapse_body_buffer(conn, responses)
   end
 
   defp add_body_to_buffer(conn, data) do
     update_in(conn.request.data_buffer, &[&1 | data])
   end
 
   defp collapse_body_buffer(conn, responses) do
     case IO.iodata_to_binary(conn.request.data_buffer) do
       "" ->
         {conn, responses}
 
       data ->
         conn = put_in(conn.request.data_buffer, [])
         {conn, [{:data, conn.request.ref, data} | responses]}
     end
   end
 
   defp store_header(%{content_length: nil} = request, "content-length", value) do
     with {:ok, content_length} <- Parse.content_length_header(value),
          do: {:ok, %{request | content_length: content_length}}
   end
 
   defp store_header(%{connection: connection} = request, "connection", value) do
     with {:ok, connection_header} <- Parse.connection_header(value),
          do: {:ok, %{request | connection: connection ++ connection_header}}
   end
 
   defp store_header(%{transfer_encoding: transfer_encoding} = request, "transfer-encoding", value) do
     with {:ok, transfer_encoding_header} <- Parse.transfer_encoding_header(value),
          do: {:ok, %{request | transfer_encoding: transfer_encoding ++ transfer_encoding_header}}
   end
 
   defp store_header(_request, "content-length", _value) do
     {:error, :more_than_one_content_length_header}
   end
 
   defp store_header(request, _name, _value) do
     {:ok, request}
   end
 
   defp request_done(%{request: request} = conn) do
     conn = pop_request(conn)
 
     cond do
       !request -> conn
       "close" in request.connection -> internal_close(conn)
       request.version >= {1, 1} -> conn
       "keep-alive" in request.connection -> conn
       true -> internal_close(conn)
     end
   end
 
   defp pop_request(conn) do
     case :queue.out(conn.requests) do
       {{:value, request}, requests} ->
         %{conn | request: request, requests: requests}
 
       {:empty, requests} ->
         %{conn | request: nil, requests: requests}
     end
   end
 
   defp enqueue_request(%{request: nil} = conn, request) do
     %__MODULE__{conn | request: request}
   end
 
   defp enqueue_request(conn, request) do
     requests = :queue.in(request, conn.requests)
     %__MODULE__{conn | requests: requests}
   end
 
   defp internal_close(conn) do
     if conn.buffer != "" do
       _ = Logger.debug(["Connection closed with data left in the buffer: ", inspect(conn.buffer)])
     end
 
     _ = conn.transport.close(conn.socket)
     %{conn | state: :closed}
   end
 
   # RFC7230 3.3.3:
   # > If a message is received with both a Transfer-Encoding and a
   # > Content-Length header field, the Transfer-Encoding overrides the
   # > Content-Length.  Such a message might indicate an attempt to
   # > perform request smuggling (Section 9.5) or response splitting
   # > (Section 9.4) and ought to be handled as an error.  A sender MUST
   # > remove the received Content-Length field prior to forwarding such
   # > a message downstream.
   defp message_body(%{body: nil, method: method, status: status} = request) do
     cond do
       status == 101 ->
         {:ok, :single}
 
       method == "HEAD" or status in 100..199 or status in [204, 304] ->
         {:ok, :none}
 
       # method == "CONNECT" and status in 200..299 -> nil
 
       request.transfer_encoding != [] && request.content_length ->
         {:error, :transfer_encoding_and_content_length}
 
       "chunked" == List.first(request.transfer_encoding) ->
         {:ok, {:chunked, nil}}
 
       request.content_length ->
         {:ok, {:content_length, request.content_length}}
 
       true ->
         {:ok, :until_closed}
     end
   end
 
   defp message_body(%{body: body}) do
     {:ok, body}
   end
 
   defp new_request(ref, method, body, encoding) do
     state =
       if body == :stream do
         {:stream_request, encoding}
       else
         :status
       end
 
     %{
       ref: ref,
       state: state,
       method: method,
       version: nil,
       status: nil,
       headers_buffer: [],
       data_buffer: [],
       content_length: nil,
       connection: [],
       transfer_encoding: [],
       body: nil
     }
   end
 
   defp lower_header_keys(headers) do
     for {name, value} <- headers, do: {Util.downcase_ascii(name), value}
   end
 
   defp add_default_headers(headers, conn) do
     headers
     |> Util.put_new_header("user-agent", @user_agent)
     |> Util.put_new_header("host", default_host_header(conn))
   end
 
   # If the port is the default for the scheme, don't add it to the host header
   defp default_host_header(%__MODULE__{scheme_as_string: scheme, host: host, port: port}) do
     if URI.default_port(scheme) == port do
       host
     else
       "#{host}:#{port}"
     end
   end
 
   defp add_content_length_or_transfer_encoding(headers, :stream) do
     cond do
       List.keymember?(headers, "content-length", 0) ->
         {:ok, headers, :identity}
 
       found = List.keyfind(headers, "transfer-encoding", 0) ->
         {"transfer-encoding", value} = found
 
         with {:ok, tokens} <- Parse.transfer_encoding_header(value) do
           if "chunked" in tokens or "identity" in tokens do
             {:ok, headers, :identity}
           else
             new_transfer_encoding = {"transfer-encoding", value <> ",chunked"}
             headers = List.keyreplace(headers, "transfer-encoding", 0, new_transfer_encoding)
             {:ok, headers, :chunked}
           end
         end
 
       # If no content-length or transfer-encoding are present, assume
       # chunked transfer-encoding and handle the encoding ourselves.
       true ->
         headers = Util.put_new_header(headers, "transfer-encoding", "chunked")
         {:ok, headers, :chunked}
     end
   end
 
   defp add_content_length_or_transfer_encoding(headers, nil) do
     {:ok, headers, :identity}
   end
 
   defp add_content_length_or_transfer_encoding(headers, body) do
     length_fun = fn -> body |> IO.iodata_length() |> Integer.to_string() end
     {:ok, Util.put_new_header_lazy(headers, "content-length", length_fun), :identity}
   end
 
   defp wrap_error(reason) do
     %HTTPError{reason: reason, module: __MODULE__}
   end
 
   @doc false
   def format_error(reason)
 
   def format_error(:closed) do
     "the connection is closed"
   end
 
   def format_error(:request_body_is_streaming) do
     "a request body is currently streaming, so no new requests can be issued"
   end
 
   def format_error({:unexpected_data, data}) do
     "received unexpected data: " <> inspect(data)
   end
 
   def format_error(:invalid_status_line) do
     "invalid status line"
   end
 
   def format_error(:invalid_header) do
     "invalid header"
   end
 
   def format_error({:invalid_request_target, target}) do
     "invalid request target: #{inspect(target)}"
   end
 
   def format_error({:invalid_header_name, name}) do
     "invalid header name: #{inspect(name)}"
   end
 
   def format_error({:invalid_header_value, name, value}) do
     "invalid value for header (only printable ASCII characters are allowed) " <>
       "#{inspect(name)}: #{inspect(value)}"
   end
 
   def format_error(:invalid_chunk_size) do
     "invalid chunk size"
   end
 
   def format_error(:missing_crlf_after_chunk) do
     "missing CRLF after chunk"
   end
 
   def format_error(:invalid_trailer_header) do
     "invalid trailer header"
   end
 
   def format_error(:more_than_one_content_length_header) do
     "the response contains two or more Content-Length headers"
   end
 
   def format_error(:transfer_encoding_and_content_length) do
     "the response contained both a Transfer-Encoding header as well as a Content-Length header"
   end
 
   def format_error({:invalid_content_length_header, value}) do
     "invalid Content-Length header: #{inspect(value)}"
   end
 
   def format_error(:empty_token_list) do
     "header should contain a list of values, but it doesn't"
   end
 
   def format_error({:invalid_token_list, string}) do
     "header contains invalid tokens: #{inspect(string)}"
   end
 
   def format_error(:trailing_headers_but_not_chunked_encoding) do
     "trailing headers can only be sent when using chunked transfer-encoding"
   end
 
   def format_error({:unallowed_trailing_header, {name, value}}) do
     "header #{inspect(name)} (with value #{inspect(value)}) is not allowed as a trailing header"
   end
 end