feat: support function streams in multipart handling

Signed-off-by: Yordis Prieto <[email protected]>

fixes #648

Author: yordis

lib/tesla/multipart.ex

@@ -33,7 +33,7 @@ defmodule Tesla.Multipart do
   end
 
   @type part_stream :: Enum.t()
-  @type part_value :: iodata | part_stream
+  @type part_value :: iodata | part_stream | function()
 
   defstruct parts: [],
             boundary: nil,
@@ -181,12 +181,13 @@ defmodule Tesla.Multipart do
 
   @spec assert_part_value!(any) :: :ok | no_return
   defp assert_part_value!(%maybe_stream{})
-       when maybe_stream in [IO.Stream, File.Stream, Stream],
+       when maybe_stream in [IO.Stream, File.Stream, Stream, Range],
        do: :ok
 
   defp assert_part_value!(value)
        when is_list(value)
-       when is_binary(value),
+       when is_binary(value)
+       when is_function(value),
        do: :ok
 
   defp assert_part_value!(val) do

test/tesla/multipart_test.exs

@@ -290,5 +290,206 @@ defmodule Tesla.MultipartTest do
 
       assert is_function(Multipart.body(mp))
     end
+
+    test "function/2 streaming response" do
+      stream_fun =
+        Stream.resource(
+          fn -> ["chunk1", "chunk2", "final"] end,
+          fn
+            [head | tail] -> {[head], tail}
+            [] -> {:halt, []}
+          end,
+          fn _ -> :ok end
+        )
+
+      mp =
+        Multipart.new()
+        |> Multipart.add_field("stream_field", stream_fun)
+
+      assert is_function(Multipart.body(mp))
+    end
+
+    test "function/2 streaming response with file content" do
+      stream_fun =
+        Stream.resource(
+          fn -> ["file data chunk 1", "file data chunk 2"] end,
+          fn
+            [head | tail] -> {[head], tail}
+            [] -> {:halt, []}
+          end,
+          fn _ -> :ok end
+        )
+
+      mp =
+        Multipart.new()
+        |> Multipart.add_file_content(stream_fun, "streamed_file.mp4")
+        |> Multipart.add_field("model", "whisper-1")
+        |> Multipart.add_field("response_format", "json")
+
+      assert is_function(Multipart.body(mp))
+
+      body_stream = Multipart.body(mp)
+      body_content = body_stream |> Enum.to_list() |> IO.iodata_to_binary()
+
+      assert body_content =~ ~s(name="file"; filename="streamed_file.mp4")
+      assert body_content =~ "file data chunk 1"
+      assert body_content =~ "file data chunk 2"
+      assert body_content =~ ~s(name="model")
+      assert body_content =~ "whisper-1"
+      assert body_content =~ ~s(name="response_format")
+      assert body_content =~ "json"
+    end
+
+    test "raw function/2 like Tesla adapter returns" do
+      stream_fun = fn
+        {:cont, acc} -> {:suspended, "data", acc}
+        {:halt, acc} -> {:halted, acc}
+      end
+
+      mp =
+        Multipart.new()
+        |> Multipart.add_file_content(stream_fun, "test.mp4")
+
+      assert %Multipart{} = mp
+      assert length(mp.parts) == 1
+    end
+  end
+
+  describe "streaming function support" do
+    @http "http://localhost:#{Application.compile_env(:httparrot, :http_port)}"
+
+    defp call_adapter(adapter, env, opts \\ []) do
+      case adapter do
+        {adapter_module, adapter_opts} ->
+          adapter_module.call(env, Keyword.merge(opts, adapter_opts))
+
+        adapter_module ->
+          adapter_module.call(env, opts)
+      end
+    end
+
+    test "accepts Stream.resource functions from Tesla adapters" do
+      stream_data =
+        Stream.resource(
+          fn -> ["chunk1", "chunk2", "chunk3"] end,
+          fn
+            [head | tail] -> {[head], tail}
+            [] -> {:halt, []}
+          end,
+          fn _ -> :ok end
+        )
+
+      mp =
+        Multipart.new()
+        |> Multipart.add_file_content(stream_data, "streamed_file.txt")
+        |> Multipart.add_field("type", "stream_test")
+
+      request = %Tesla.Env{
+        method: :post,
+        url: "#{@http}/post",
+        body: mp
+      }
+
+      assert {:ok, response} = call_adapter(Tesla.Adapter.Mint, request)
+      assert response.status == 200
+
+      {:ok, response} = Tesla.Middleware.JSON.decode(response, [])
+
+      assert response.body["form"]["type"] == "stream_test"
+      assert response.body["files"]["file"] == "chunk1chunk2chunk3"
+
+      content_type = response.body["headers"]["content-type"]
+      assert content_type =~ "multipart/form-data"
+      assert content_type =~ "boundary="
+    end
+
+    test "works with File.Stream across different adapters" do
+      file_stream = File.stream!("test/tesla/multipart_test_file.sh")
+
+      mp =
+        Multipart.new()
+        |> Multipart.add_file_content(file_stream, "test_script.sh")
+        |> Multipart.add_field("adapter", "hackney")
+
+      request = %Tesla.Env{
+        method: :post,
+        url: "#{@http}/post",
+        body: mp
+      }
+
+      assert {:ok, response} = call_adapter(Tesla.Adapter.Hackney, request)
+      assert response.status == 200
+
+      {:ok, response} = Tesla.Middleware.JSON.decode(response, [])
+
+      assert response.body["form"]["adapter"] == "hackney"
+
+      assert response.body["files"]["file"] ==
+               "#!/usr/bin/env bash\necho \"test multipart file\"\n"
+    end
+
+    test "reproduces GitHub issue #648 scenario" do
+      gcp_stream_response =
+        Stream.resource(
+          fn -> ["audio_data_chunk_1", "audio_data_chunk_2", "audio_data_chunk_3"] end,
+          fn
+            [head | tail] -> {[head], tail}
+            [] -> {:halt, []}
+          end,
+          fn _ -> :ok end
+        )
+
+      upload_body =
+        Multipart.new()
+        |> Multipart.add_file_content(gcp_stream_response, "audio.mp4")
+        |> Multipart.add_field("model", "whisper-1")
+        |> Multipart.add_field("response_format", "json")
+
+      request = %Tesla.Env{
+        method: :post,
+        url: "#{@http}/post",
+        body: upload_body
+      }
+
+      assert {:ok, response} = call_adapter(Tesla.Adapter.Mint, request)
+      assert response.status == 200
+
+      {:ok, response} = Tesla.Middleware.JSON.decode(response, [])
+
+      assert response.body["form"]["model"] == "whisper-1"
+      assert response.body["form"]["response_format"] == "json"
+
+      assert response.body["files"]["file"] ==
+               "audio_data_chunk_1audio_data_chunk_2audio_data_chunk_3"
+    end
+
+    test "handles large streaming data efficiently" do
+      large_stream = Stream.repeatedly(fn -> "data_chunk_" end) |> Stream.take(100)
+
+      mp =
+        Multipart.new()
+        |> Multipart.add_file_content(large_stream, "large_file.dat")
+        |> Multipart.add_field("size", "large")
+
+      request = %Tesla.Env{
+        method: :post,
+        url: "#{@http}/post",
+        body: mp
+      }
+
+      assert {:ok, response} = call_adapter(Tesla.Adapter.Mint, request)
+      assert response.status == 200
+
+      {:ok, response} = Tesla.Middleware.JSON.decode(response, [])
+
+      assert response.body["form"]["size"] == "large"
+
+      file_content = response.body["files"]["file"]
+      assert String.contains?(file_content, "data_chunk_")
+
+      expected_length = String.length("data_chunk_") * 100
+      actual_length = String.length(file_content)
+      assert abs(actual_length - expected_length) < 50
+    end
   end
 end