http_stream.exs

Mix.install([:mint, :castore])

defmodule Main do
  def run() do
    total =
      Stream.resource(
        fn ->
          {:ok, conn} = Mint.HTTP.connect(:https, "ftp.bit.nl", 443, mode: :passive)
          {:ok, conn, _ref} = Mint.HTTP.request(conn, "GET", "/speedtest/10mb.bin", [], nil)
          {conn, true}
        end,
        fn
          {conn, true} ->
            case Mint.HTTP.recv(conn, 0, 5_000) do
              {:ok, conn, responses} ->
                case process_responses(responses) do
                  {:ok, binaries, cont} -> {binaries, {conn, cont}}
                  {:error, reason} -> raise "Error streaming: #{inspect(reason)}"
                end

              _ ->
                raise "Something went wrong"
            end

          {conn, false} ->
            {:halt, conn}
        end,
        fn conn ->
          Mint.HTTP.close(conn)
        end
      )
      |> Stream.each(fn chunk ->
        IO.puts("Streamed chunk with #{byte_size(chunk)} bytes")
      end)
      |> Stream.map(&byte_size/1)
      |> Enum.sum()

    IO.puts("Received #{total} bytes")
  end

  defp process_responses(response) do
    process_responses(response, [], true)
  end

  defp process_responses([], binaries, more) do
    {:ok, binaries, more}
  end

  defp process_responses([{:status, _, _status} | rest], binaries, more) do
    process_responses(rest, binaries, more)
  end

  defp process_responses([{:headers, _, _headers} | rest], binaries, more) do
    process_responses(rest, binaries, more)
  end

  defp process_responses([{:data, _, binary} | rest], binaries, more) do
    process_responses(rest, binaries ++ [binary], more)
  end

  defp process_responses([{:done, _} | rest], binaries, _more) do
    process_responses(rest, binaries, false)
  end

  defp process_responses([{:error, _, reason} | rest], _binaries, _more) do
    {:error, reason}
  end
end

Main.run()