Repository: wojtekmach/req (elixir)

Issue:
Multipart should support streaming bodies
Per https://hexdocs.pm/req/Req.Steps.html#encode_body/1 
`:form_multipart` supports only
- integer
- iodata
- File.Stream

So, this is not possible as of today.
```elixir
%{body: body} = Req.get!("...", into: self) # todo: back pressure
file = {body, content_type: "video/mp4", filename: "..."} # todo: include :size
Req.post!(form_multipart: [file: file])
```

I tried to patch the lib and surprisingly it worked. See https://github.com/wojtekmach/req/compare/main...knightpp:req:add-multipart-stream

My use case is: restream video bytes from a GET request into POST multipart.

Pull request description:
Support any enumerable in mutlipart
Closes #442

Patch:
diff --git a/lib/req/steps.ex b/lib/req/steps.ex
index d00a504..ff0d018 100644
--- a/lib/req/steps.ex
+++ b/lib/req/steps.ex
@@ -398,7 +398,9 @@ defmodule Req.Steps do
 
         * `File.Stream`
 
-        * `{value, options}` tuple. Supported options are `:filename` and `:content_type`
+        * `Enumerable`
+
+        * `{value, options}` tuple. Supported options are: `:filename`, `:content_type`, `:content_length`
 
     * `:json` - if set, encodes the request body as JSON (using `Jason.encode_to_iodata!/1`), sets
       the `accept` header to `application/json`, and the `content-type` header to `application/json`.
@@ -419,6 +421,14 @@ defmodule Req.Steps do
       iex> resp.body["files"]
       %{"b" => "2"}
 
+  Encoding streaming form (`multipart/form-data`):
+
+      iex> stream = Stream.cycle(["abc"]) |> Stream.take(3)
+      iex> fields = [file: {stream, filename: "b.txt", content_length: 9}]
+      iex> resp = Req.post!("https://httpbin.org/anything", form_multipart: fields)
+      iex> resp.body["files"]
+      %{"file" => "abcabcabc"}
+
   Encoding JSON:
 
       iex> Req.post!("https://httpbin.org/post", json: %{a: 2}).body["json"]
diff --git a/lib/req/utils.ex b/lib/req/utils.ex
index 1dfc52f..343153c 100644
--- a/lib/req/utils.ex
+++ b/lib/req/utils.ex
@@ -493,6 +493,11 @@ defmodule Req.Utils do
     }
   end
 
+  defp add_form_parts({_parts1, size1}, {_parts2, size2})
+       when not is_integer(size1) or not is_integer(size2) do
+    raise ArgumentError, "multipart part sizes must be integers"
+  end
+
   defp add_form_parts({parts1, size1}, {parts2, size2})
        when is_list(parts1) and is_list(parts2) do
     {[parts1, parts2], size1 + size2}
@@ -503,7 +508,7 @@ defmodule Req.Utils do
   end
 
   defp encode_form_part({name, {value, options}}, boundary) do
-    options = Keyword.validate!(options, [:filename, :content_type])
+    options = Keyword.validate!(options, [:filename, :content_type, :content_length])
 
     {parts, parts_size, options} =
       case value do
@@ -533,6 +538,12 @@ defmodule Req.Utils do
             end)
 
           {stream, size, options}
+
+        enum ->
+          Enumerable.impl_for!(enum)
+          size = Keyword.get(options, :content_length, 0)
+
+          {enum, size, options}
       end
 
     params =
@@ -543,11 +554,11 @@ defmodule Req.Utils do
       end
 
     headers =
-      if content_type = options[:content_type] do
-        ["content-type: ", content_type, @crlf]
-      else
-        []
-      end
+      [
+        maybe_content_type(options),
+        maybe_content_length(options)
+      ]
+      |> Enum.reject(&is_nil/1)
 
     headers = ["content-disposition: form-data; name=\"#{name}\"", params, @crlf, headers]
     header = [[@crlf, "--", boundary, @crlf, headers, @crlf]]
@@ -558,6 +569,18 @@ defmodule Req.Utils do
     encode_form_part({name, {value, []}}, boundary)
   end
 
+  defp maybe_content_type(options) do
+    if content_type = options[:content_type] do
+      ["content-type: ", content_type, @crlf]
+    end
+  end
+
+  defp maybe_content_length(options) do
+    if content_length = options[:content_length] do
+      ["content-length: ", Integer.to_string(content_length), @crlf]
+    end
+  end
+
   @doc """
   Loads .netrc file.
 
diff --git a/mix.exs b/mix.exs
index 2569382..91790e0 100644
--- a/mix.exs
+++ b/mix.exs
@@ -16,11 +16,6 @@ defmodule Req.MixProject do
       aliases: [
         "test.all": ["test --include integration"]
       ],
-      preferred_cli_env: [
-        "test.all": :test,
-        docs: :docs,
-        "hex.publish": :docs
-      ],
       xref: [
         exclude: [
           NimbleCSV.RFC4180,
@@ -40,6 +35,16 @@ defmodule Req.MixProject do
     ]
   end
 
+  def cli do
+    [
+      preferred_envs: [
+        "test.all": :test,
+        docs: :docs,
+        "hex.publish": :docs
+      ]
+    ]
+  end
+
   defp package do
     [
       description: "Req is a batteries-included HTTP client for Elixir.",


Test patch:
diff --git a/test/req/httpc_test.exs b/test/req/httpc_test.exs
index 39b896d..97e59ed 100644
--- a/test/req/httpc_test.exs
+++ b/test/req/httpc_test.exs
@@ -1,135 +1,7 @@
 # Experimental httpc adapter to test the adapter contract.
 
-defmodule Req.HttpcTest do
-  use ExUnit.Case, async: true
-
-  require Logger
-
-  # TODO
-  @moduletag :skip
-
-  setup do
-    bypass = Bypass.open()
-
-    req =
-      Req.new(
-        adapter: &run_httpc/1,
-        url: "http://localhost:#{bypass.port}"
-      )
-
-    [bypass: bypass, req: req]
-  end
-
-  if function_exported?(Mix, :ensure_application!, 1) do
-    Mix.ensure_application!(:inets)
-  end
-
-  describe "run_httpc/1" do
-    test "request", %{bypass: bypass, req: req} do
-      Bypass.expect(bypass, "GET", "/", fn conn ->
-        Plug.Conn.send_resp(conn, 200, "ok")
-      end)
-
-      resp = Req.get!(req)
-      assert resp.status == 200
-      assert Req.Response.get_header(resp, "server") == ["Cowboy"]
-      assert resp.body == "ok"
-    end
-
-    test "post request body", %{bypass: bypass, req: req} do
-      Bypass.expect(bypass, "POST", "/", fn conn ->
-        assert {:ok, body, conn} = Plug.Conn.read_body(conn)
-        Plug.Conn.send_resp(conn, 200, body)
-      end)
-
-      resp = Req.post!(req, body: "foofoofoo")
-      assert resp.status == 200
-      assert resp.body == "foofoofoo"
-    end
-
-    test "stream request body", %{bypass: bypass, req: req} do
-      Bypass.expect(bypass, "POST", "/", fn conn ->
-        assert {:ok, body, conn} = Plug.Conn.read_body(conn)
-        Plug.Conn.send_resp(conn, 200, body)
-      end)
-
-      resp = Req.post!(req, body: {:stream, Stream.take(["foo", "foo", "foo"], 2)})
-      assert resp.status == 200
-      assert resp.body == "foofoo"
-    end
-
-    test "into: fun", %{req: req, bypass: bypass} do
-      Bypass.expect(bypass, "GET", "/", fn conn ->
-        conn = Plug.Conn.send_chunked(conn, 200)
-        {:ok, conn} = Plug.Conn.chunk(conn, "foo")
-        {:ok, conn} = Plug.Conn.chunk(conn, "bar")
-        conn
-      end)
-
-      pid = self()
-
-      resp =
-        Req.get!(
-          req,
-          into: fn {:data, data}, acc ->
-            send(pid, {:data, data})
-            {:cont, acc}
-          end
-        )
-
-      assert resp.status == 200
-      assert resp.headers["transfer-encoding"] == ["chunked"]
-      assert_receive {:data, "foobar"}
-
-      # httpc seems to randomly chunk things
-      receive do
-        {:data, ""} -> :ok
-      after
-        0 -> :ok
-      end
-
-      refute_receive _
-    end
-
-    test "into: :self", %{req: req, bypass: bypass} do
-      Bypass.expect(bypass, "GET", "/", fn conn ->
-        conn = Plug.Conn.send_chunked(conn, 200)
-        {:ok, conn} = Plug.Conn.chunk(conn, "foo")
-        {:ok, conn} = Plug.Conn.chunk(conn, "bar")
-        conn
-      end)
-
-      resp = Req.get!(req, into: :self)
-      assert resp.status == 200
-
-      # httpc seems to randomly chunk things
-      assert Req.parse_message(resp, assert_receive(_)) in [
-               {:ok, [data: "foo"]},
-               {:ok, [data: "foobar"]}
-             ]
-
-      assert Req.parse_message(resp, assert_receive(_)) in [
-               {:ok, [data: "bar"]},
-               {:ok, [data: ""]},
-               {:ok, [:done]}
-             ]
-    end
-
-    test "into: pid cancel", %{req: req, bypass: bypass} do
-      Bypass.expect(bypass, "GET", "/", fn conn ->
-        conn = Plug.Conn.send_chunked(conn, 200)
-        {:ok, conn} = Plug.Conn.chunk(conn, "foo")
-        {:ok, conn} = Plug.Conn.chunk(conn, "bar")
-        conn
-      end)
-
-      resp = Req.get!(req, into: :self)
-      assert resp.status == 200
-      assert :ok = Req.cancel_async_response(resp)
-    end
-  end
-
-  def run_httpc(request) do
+defmodule Req.Httpc do
+  def run(request) do
     httpc_url = request.url |> URI.to_string() |> String.to_charlist()
 
     httpc_headers =
@@ -331,3 +203,133 @@ defmodule Req.HttpcTest do
     end
   end
 end
+
+defmodule Req.HttpcTest do
+  use ExUnit.Case, async: true
+
+  require Logger
+
+  # TODO
+  @moduletag :skip
+
+  setup do
+    bypass = Bypass.open()
+
+    req =
+      Req.new(
+        adapter: &Req.Httpc.run/1,
+        url: "http://localhost:#{bypass.port}"
+      )
+
+    [bypass: bypass, req: req]
+  end
+
+  if function_exported?(Mix, :ensure_application!, 1) do
+    Mix.ensure_application!(:inets)
+  end
+
+  describe "httpc" do
+    test "request", %{bypass: bypass, req: req} do
+      Bypass.expect(bypass, "GET", "/", fn conn ->
+        Plug.Conn.send_resp(conn, 200, "ok")
+      end)
+
+      resp = Req.get!(req)
+      assert resp.status == 200
+      assert Req.Response.get_header(resp, "server") == ["Cowboy"]
+      assert resp.body == "ok"
+    end
+
+    test "post request body", %{bypass: bypass, req: req} do
+      Bypass.expect(bypass, "POST", "/", fn conn ->
+        assert {:ok, body, conn} = Plug.Conn.read_body(conn)
+        Plug.Conn.send_resp(conn, 200, body)
+      end)
+
+      resp = Req.post!(req, body: "foofoofoo")
+      assert resp.status == 200
+      assert resp.body == "foofoofoo"
+    end
+
+    test "stream request body", %{bypass: bypass, req: req} do
+      Bypass.expect(bypass, "POST", "/", fn conn ->
+        assert {:ok, body, conn} = Plug.Conn.read_body(conn)
+        Plug.Conn.send_resp(conn, 200, body)
+      end)
+
+      resp = Req.post!(req, body: {:stream, Stream.take(["foo", "foo", "foo"], 2)})
+      assert resp.status == 200
+      assert resp.body == "foofoo"
+    end
+
+    test "into: fun", %{req: req, bypass: bypass} do
+      Bypass.expect(bypass, "GET", "/", fn conn ->
+        conn = Plug.Conn.send_chunked(conn, 200)
+        {:ok, conn} = Plug.Conn.chunk(conn, "foo")
+        {:ok, conn} = Plug.Conn.chunk(conn, "bar")
+        conn
+      end)
+
+      pid = self()
+
+      resp =
+        Req.get!(
+          req,
+          into: fn {:data, data}, acc ->
+            send(pid, {:data, data})
+            {:cont, acc}
+          end
+        )
+
+      assert resp.status == 200
+      assert resp.headers["transfer-encoding"] == ["chunked"]
+      assert_receive {:data, "foobar"}
+
+      # httpc seems to randomly chunk things
+      receive do
+        {:data, ""} -> :ok
+      after
+        0 -> :ok
+      end
+
+      refute_receive _
+    end
+
+    test "into: :self", %{req: req, bypass: bypass} do
+      Bypass.expect(bypass, "GET", "/", fn conn ->
+        conn = Plug.Conn.send_chunked(conn, 200)
+        {:ok, conn} = Plug.Conn.chunk(conn, "foo")
+        {:ok, conn} = Plug.Conn.chunk(conn, "bar")
+        conn
+      end)
+
+      resp = Req.get!(req, into: :self)
+      assert resp.status == 200
+
+      # httpc seems to randomly chunk things
+      assert Req.parse_message(resp, assert_receive(_)) in [
+               {:ok, [data: "foo"]},
+               {:ok, [data: "foobar"]}
+             ]
+
+      assert Req.parse_message(resp, assert_receive(_)) in [
+               {:ok, [data: "bar"]},
+               {:ok, [data: ""]},
+               {:ok, [:done]}
+             ]
+    end
+
+    test "into: pid cancel", %{req: req, bypass: bypass} do
+      Bypass.expect(bypass, "GET", "/", fn conn ->
+        conn = Plug.Conn.send_chunked(conn, 200)
+        {:ok, conn} = Plug.Conn.chunk(conn, "foo")
+        {:ok, conn} = Plug.Conn.chunk(conn, "bar")
+        conn
+      end)
+
+      resp = Req.get!(req, into: :self)
+      assert resp.status == 200
+      assert :ok = Req.cancel_async_response(resp)
+    end
+  end
+end
diff --git a/test/req/utils_test.exs b/test/req/utils_test.exs
index 43db64d..26cc58b 100644
--- a/test/req/utils_test.exs
+++ b/test/req/utils_test.exs
@@ -203,6 +203,45 @@ defmodule Req.UtilsTest do
              """
     end
 
+    test "raise when size is not integer" do
+      assert_raise ArgumentError, fn ->
+        enum = Stream.cycle([1]) |> Stream.take(1)
+        Req.Utils.encode_form_multipart([field1: {enum, content_length: "10"}], boundary: "foo")
+      end
+    end
+
+    test "content-length" do
+      %{content_type: content_type, body: body, size: size} =
+        Req.Utils.encode_form_multipart([field1: {"a", content_length: 11}], boundary: "foo")
+
+      body = IO.iodata_to_binary(body)
+
+      assert size == byte_size(body)
+      assert content_type == "multipart/form-data; boundary=foo"
+
+      assert body == """
+             \r\n\
+             --foo\r\n\
+             content-disposition: form-data; name=\"field1\"\r\n\
+             content-length: 11\r\n\
+             \r\n\
+             a\r\n\
+             --foo--\r\n\
+             """
+    end
+
+    test "can accept any enumerable" do
+      enum = Stream.cycle(["a"]) |> Stream.take(1)
+
+      %{body: body, size: size} =
+        Req.Utils.encode_form_multipart([field1: {enum, content_length: 11}], boundary: "foo")
+
+      body = body |> Enum.to_list() |> IO.iodata_to_binary()
+
+      # content_length is +10 of actual size
+      assert size == byte_size(body) + 10
+    end
+
     @tag :tmp_dir
     test "can return stream", %{tmp_dir: tmp_dir} do
       File.write!("#{tmp_dir}/2.txt", "22")