Handling http response streaming

I have never actually done this but my impression reading the docs is that you’re meant to use the response in the {req, resp} tuple (your context) as an accumulator. You could place the chunks in the :private field of resp under your own key. See :private under “Fields” at Req.Response.

Or you could use the response :body - not sure if it would get clobbered by anything else.

It would be nice if the docs for :into actually spelled this out as I was looking into this exact thing a couple weeks ago and had to infer that this was the intended approach. Seems like it would be a very common use case.

Yeah I saw the :private field but was scared off with the detail

a map reserved for libraries and frameworks to use. Prefix the keys with the name of your project to avoid any future conflicts. Only accepts atom/0 keys.

But it’s worth a shot

Yeah, sorry about that, it’s on my roadmap to improve documentation around this. req.private and resp.private exist solely so when writing steps and into: fun we can store intermediate state. The only restriction is private.req_* key names are reserved for Req for forward-compatibility.

ok. so I should be able to use resp.private to append the chunks as they come back into a list. sweet. I’ll try it out and post the results. I tried with an Agent and managed to get the results back correctly, probably not idiomatic Elixir.

  def transcribe_audio(url, callback) do
    req =
      Req.new(
        url: "http://faster-whisper-server-patient-voice-8559.flycast/v1/audio/transcriptions",
        connect_options: [transport_opts: [inet6: true]],
        receive_timeout: 120_000
      )

    file_path =
      file_path_from_url(
        "https://s3-us-west-2.amazonaws.com/staging-moth-social/media_attachments/files/113/041/865/730/367/819/original/9fd57c861d8f3fc7.mp3"
      )

    multipart =
      Multipart.new()
      |> Multipart.add_part(Multipart.Part.text_field("true", "stream"))
      |> Multipart.add_part(Multipart.Part.file_field(file_path, "file"))

    content_length = Multipart.content_length(multipart)
    content_type = Multipart.content_type(multipart, "multipart/form-data")

    headers = [
      {"Content-Type", content_type},
      {"Content-Length", to_string(content_length)}
    ]

    # Initialize buffer state
    {:ok, agent} = Agent.start_link(fn -> [] end)

    Req.post(
      req,
      headers: headers,
      body: Multipart.body_stream(multipart),
      into: fn {:data, data}, context ->
        # fn {:ok, {ss, %{chunks: [%{text: text}]}}} -> func.(ss, text) end
        results = parse(data)
        segment = Enum.at(results.segments, 0)
        callback.(trunc(segment.start), segment.text)
        :ok = Agent.update(agent, fn state -> [results | state] end)
        {:cont, context}
      end
    )

    # Make sure we shut the agent down
    data_results = Agent.get(agent, & &1) |> Enum.reverse()
    :ok = Agent.stop(agent)
    data_results
  end

I don’t seem to be able to append/update :private from within the into: fun

 def transcribe_audio(url, callback) do
    req =
      Req.new(
        url: "http://faster-whisper-server-patient-voice-8559.flycast/v1/audio/transcriptions",
        connect_options: [transport_opts: [inet6: true]],
        receive_timeout: 120_000
      )

    file_path =
      file_path_from_url(
        "https://s3-us-west-2.amazonaws.com/moth-social/media_attachments/files/113/087/363/649/463/883/original/6f9b1ad7ff1c0018.mp3"
      )

    multipart =
      Multipart.new()
      |> Multipart.add_part(Multipart.Part.text_field("true", "stream"))
      |> Multipart.add_part(Multipart.Part.file_field(file_path, "file"))

    content_length = Multipart.content_length(multipart)
    content_type = Multipart.content_type(multipart, "multipart/form-data")

    headers = [
      {"Content-Type", content_type},
      {"Content-Length", to_string(content_length)}
    ]

    {:ok, response} =
      Req.post(
        req,
        headers: headers,
        body: Multipart.body_stream(multipart),
        into: fn {:data, data}, {req, resp} ->
          # try to add any value to the response
          Req.Response.put_private(resp, :transcription, [{%{:b => "moe"}}])
          results = parse(data)
          segment = Enum.at(results.segments, 0)
          callback.(trunc(segment.start), segment.text)
          # update private response
          Req.Response.update_private(resp, :transcript, [], fn state -> [results | state] end)
          IO.inspect(resp.private)
          {:cont, {req, resp}}
        end
      )

    Req.Response.put_private(response, :something_else, [{%{:a => "ted"}}])

    # Make sure we shut the agent down
    # data_results = Agent.get(agent, & &1) |> Enum.reverse()
    # :ok = Agent.stop(agent)
    # data_results
  end

the put_private or update_private from the into: fun are not on the return response, but the final put_private is added

#Req.Response<
  status: 200,
  headers: %{
    "content-type" => ["text/event-stream; charset=utf-8"],
    "date" => ["Thu, 05 Sep 2024 23:39:26 GMT"],
    "fly-request-id" => ["01J728X25018F47F3W0KD31EJ1-ord"],
    "server" => ["Fly/813cb6e67 (2024-09-03)"],
    "transfer-encoding" => ["chunked"],
    "via" => ["1.1 fly.io"]
  },
  body: "",
  trailers: %{},
  private: %{something_else: [{%{a: "ted"}}]},
  ...
>

put_private, update_private, etc return updated response, you need to reassign the resp variable.

Btw you can nowadays drop Multipart dependency since we have this capability built in, see encode_body/1 step.

d’oh. There we go. Thank you for your help, and the tip on multi part.