About Req and redirections and progress download

[ndrean]

This is a digression from the main ML topic but I was curious about Req and the code you provided for redirections. Since Req is based on Finch, I provided the Finch code as well.

In the code below, the server triggers an HTTP request to an endpoint and streams the response and writes by chunk into a file.
If you want to save the data into a file, you may need to use IO.binwrite to append the chunks into the file process, straight from the Finch doc.

Redirections are allowed and Req follows the redirection. - without doing anything.

I also added a progress count. The progress is saved into the %Req.Response{} under the :private key.

[Front-end note] an HTTP request made from the browser uses the fetch API which can consume a fetch as a stream. To get a download progress, you can use fetch and set a RedableStream on the response body (res.body.getReader()). However, it is easier to use XMLHttpRequest (because you get an accumulator for free with onprogress). For an upload progress, you need to use XMLHttpRequest. This is the reason why Liveview used it in the uploader snippet.

The code below is "livebookable" 😁

==================================================================================================

HTTP streaming: server download with progress with Finch and Req

Mix.install([
  {:finch, "~> 0.18.0"},
  {:req, "~> 0.4.14"}
])

Finch.start_link(name: ExStream.Finch)

Stream & progress with Finch

We start with the Finch library.

defmodule FinchStream do
  @doc """
  Read or stream download the file server-side
  """
  def get_file(url, file_path) do
    case File.ls!() |> Enum.member?(file_path) do
      true ->
        {:ok, file_path}

      false ->
        download(url, file_path)
    end
  end

  @doc """
  Take an URL and save the file to the path.
  """

  def download(url, file_path) do
    IO.puts("Starting to process #{inspect(file_path)}...........")

    # Open a file to which binary chunks will be appended to.
    # this process is reset in case of redirection
    file_pid = File.open!(file_path, [:write, :binary])

    unless is_pid(file_pid), do: raise("File creation problem on disk")

    # the HTTP stream request
    Finch.build(:get, url)
    |> Finch.stream_while(ExStream.Finch, nil, fn
      # we put the status in the "acc" to handle redirections
      {:status, status}, _acc ->
        {:cont, status}

      # - when we receive 302, we put the "location" header in the "acc"
      # - when we receive a 200, we put the "content-length" and the file name in the "acc",
      {:headers, headers}, acc ->
        handle_headers(headers, acc)

      # when we receive the "location" tuple, we recurse
      # otherwise, we write the chunk into the file and print out the current progress.
      {:data, data}, acc ->
        handle_data(data, acc, file_path, file_pid)
    end)

    case File.close(file_pid) do
      :ok ->
        {:halt, {file_path, :done}}

      {:error, _reason} ->
        {:halt, :error}
    end
  end

  def handle_headers(headers, status) when status in [301, 302, 303, 307, 308] do
    IO.puts("#{status}: REDIR")
   
    {:cont, Enum.find(headers, &(elem(&1, 0) == "location"))}
  end

  def handle_headers(headers, 200) do
    {"content-length", size} =
      Enum.find(headers, &(elem(&1, 0) == "content-length"))

    case size do
      nil ->
        {:cont, {0, nil}}

      size  ->
        {:cont, {0, String.to_integer(size)}}
    end
  end

  def handle_headers(_, status) do  
    dbg(status)
    {:halt, :bad_status}
  end

  def handle_data(_data, {"location", location}, file_path, file_pid) do
    if Process.alive?(file_pid), do:
       :ok = File.close(file_pid)

    download(location, file_path)
  end

  def handle_data(data, {processed, size}, file_path, file_pid) do
    case IO.binwrite(file_pid, data) do
      :ok ->
        processed =
          if is_integer(size) and size > 0 do
            (processed + byte_size(data))
            |> tap(fn processed ->
              IO.inspect(Float.round(processed * 100 / size, 1),
                label: "Processed #{inspect(file_path)} %: "
              )
            end)
          else
            processed + byte_size(data)
          end

        {:cont, {processed, size}}

      {:error, reason} ->
        {:error, reason}
    end
  end
end

Test endpoints

We use two endpoints for testing.

vid = "https://sample-videos.com/video321/mp4/720/big_buck_bunny_720p_1mb.mp4"
img = "https://source.unsplash.com/QT-l619id6w"

We will run concurrently two task using Task.async_stream since we use one function with different arguments.

:timer.tc(fn ->
  Task.async_stream(
    [[vid, "video.mp4"], [img, "image.jpg"]],
    &apply(FinchStream, :download, &1),
    timeout: :infinity
  )
  # |> Stream.run()
  |> Enum.to_list()
end)

Stream & progress with Req

We can download with streams and write the chunks into a file. The code is pretty compact.

defmodule ReqWriteStream do
  def req_write_stream(url, file_path) do
    Req.get!(url, raw: true, into: File.stream!(file_path, [:write]))
  end
end

:timer.tc(fn ->
  ReqWriteStream.req_write_stream(img, "image2.jpg")
end)

We write a module that takes into account the redirections and displays the progress by using the :private of the struct Req.Response{}.

defmodule ReqProgressStream do
  def req_progress_stream(url, file_path) do
    [size]  = Map.get(Req.head!(url: url).headers, "content-length", ["0"])
    size = String.to_integer(size)
    file_pid = File.open!(file_path, [:write, :binary])

    func = fn {:data, data}, {req, res} ->
      IO.binwrite(file_pid, data)
      chunk_size = byte_size(data)
      res = Req.Response.update_private(res, :progress, chunk_size, &(&1 + chunk_size))

      if size>0, do:
        {Req.Response.get_private(res, :progress) * 100 / size, chunk_size, size} |> dbg()
        
      {:cont, {req, res}}
    end

    Req.get!(url: url, raw: true, into: func)
    |> dbg()

    File.close(file_pid)
  end
end

We run the HTTP calls concurrently with Task.async and Task.await_many.

:timer.tc(fn ->
  [
    Task.async(fn -> ReqProgressStream.req_progress_stream(vid, "video2.mp4") end),
    Task.async(fn -> ReqProgressStream.req_progress_stream(img, "image3.jpg") end)
  ]
  |> Task.await_many(:infinity)
end)

We can alternatively use again Task.async_stream.

:timer.tc(fn ->
  Task.async_stream(
    [[vid, "video3.mp4"], [img, "image4.jpg"]],
    &apply(ReqProgressStream, :req_progress_stream, &1),
    timeout: :infinity
  )
  |> Stream.run()
end)