Redesign around Req.stream, step.stream, and adapter.stream

[wojtekmach]

The current design around request, response, and error steps, especially response steps, is fundamentally at odds with streaming. We cannot perform streaming decompression/decoding in response steps since by that time we've already received response body.

The other streaming solution, into: :self and %Req.Response.Async{} = resp.body is not satisfactory either as while it allows composing that enumerable, we're receiving messages to the current process with no back pressure.

We're redesigning Req around arguably the best response streaming option, Finch.stream_while/5.

In this PR we're introducing a few things:

• Req.stream(req, acc, fun, options \\ [])

• steps are now fn req -> req and mod.stream(req, acc, fun, state, next). next, familiar to Tesla users, allows calling the rest of the pipeline.

• Adapter contract is now adapter.stream(req, acc, fun, state)

Examples

OpenAI Chat Completions

{:ok, _resp, acc} =
  Req.stream(
    "https://api.openai.com/v1/chat/completions",
    %{request_id: nil, total_tokens: 0},
    fn
      %{data: "[DONE]"}, _resp, acc ->
        {:cont, acc}

      %{data: data}, resp, acc ->
        [request_id] = Req.Response.get_header(resp, "x-request-id")
        acc = put_in(acc.request_id, request_id)

        case JSON.decode!(data) do
          %{"choices" => [%{"delta" => %{"content" => chunk}} | _]} ->
            IO.write(chunk)
            {:cont, acc}

          %{"usage" => %{} = usage} ->
            {:cont, put_in(acc.total_tokens, usage["total_tokens"])}

          _other ->
            {:cont, acc}
        end
    end,
    auth: {:bearer, System.fetch_env!("OPENAI_API_KEY")},
    json: %{
      model: "gpt-4o-mini",
      messages: [%{role: "user", content: "Write a haiku about Elixir."}],
      stream: true,
      stream_options: %{include_usage: true}
    }
  )

IO.puts("\n")
dbg(acc)

NDJSON

{:ok, resp, acc} =
  Req.stream(
    "https://httpbin.org/stream/3",
    nil,
    fn data, _resp, acc ->
      dbg(data)
      {:cont, acc}
    end,
    decoders: [json: :ndjson] # httpbin.org sets content-type: application/json.
  )

Reverse Proxy

Mix.install([
  {:req, github: "wojtekmach/req", branch: "wm-stream"},
  :bandit
])

defmodule Proxy do
  def start_link(upstream_url: upstream_url) do
    with {:ok, pid} <- Bandit.start_link(port: 0, plug: {&forward/2, upstream_url}),
         {:ok, {_ip, port}} <- ThousandIsland.listener_info(pid) do
      {:ok, %{url: URI.new!("http://localhost:#{port}")}}
    end
  end

  defp forward(conn, upstream_url) do
    {:ok, _resp, conn} =
      Req.stream(
        upstream_url,
        conn,
        fn
          data, resp, %{state: :unset} = conn ->
            [content_type] = Req.Response.get_header(resp, "content-type")

            conn =
              conn
              |> Plug.Conn.put_resp_header("content-type", content_type)
              |> Plug.Conn.send_chunked(resp.status)

            {:ok, conn} = Plug.Conn.chunk(conn, data)
            {:cont, conn}

          data, _resp, %{state: :chunked} = conn ->
            {:ok, conn} = Plug.Conn.chunk(conn, data)
            {:cont, conn}
        end,
        method: conn.method,
        body: fn conn ->
          case Plug.Conn.read_body(conn) do
            {:ok, "", conn} ->
              {:done, conn}

            {:ok, data, conn} ->
              {:data, data, conn}

            {:more, data, conn} ->
              {:data, data, conn}
          end
        end,
        raw: true
      )

    conn
  end
end

{:ok, %{url: url}} = Proxy.start_link(upstream_url: "https://httpbin.org/anything")

resp = Req.post!(url, body: "Hello, World!")
dbg(resp.body)

[benjreinhart]

IIRC, we exposed the Parser API used here in 1.1.0. So I think you might want this to be ~> 1.1

[wojtekmach]

Yup, done, thanks!