Request/Response streaming for Finch adapter

Author: teamon

Diff for: lib/tesla/adapter/finch.ex

@@ -52,15 +52,20 @@ if Code.ensure_loaded?(Finch) do
     @behaviour Tesla.Adapter
     alias Tesla.Multipart
 
+    @defaults [
+      receive_timeout: 15_000
+    ]
+
     @impl Tesla.Adapter
     def call(%Tesla.Env{} = env, opts) do
-      opts = Tesla.Adapter.opts(env, opts)
+      opts = Tesla.Adapter.opts(@defaults, env, opts)
 
       name = Keyword.fetch!(opts, :name)
       url = Tesla.build_url(env.url, env.query)
       req_opts = Keyword.take(opts, [:pool_timeout, :receive_timeout])
+      req = build(env.method, url, env.headers, env.body)
 
-      case request(name, env.method, url, env.headers, env.body, req_opts) do
+      case request(req, name, req_opts, opts) do
         {:ok, %Finch.Response{status: status, headers: headers, body: body}} ->
           {:ok, %Tesla.Env{env | status: status, headers: headers, body: body}}
 
@@ -69,20 +74,72 @@ if Code.ensure_loaded?(Finch) do
       end
     end
 
-    defp request(name, method, url, headers, %Multipart{} = mp, opts) do
+    defp build(method, url, headers, %Multipart{} = mp) do
       headers = headers ++ Multipart.headers(mp)
       body = Multipart.body(mp) |> Enum.to_list()
 
-      request(name, method, url, headers, body, opts)
+      build(method, url, headers, body)
+    end
+
+    defp build(method, url, headers, %Stream{} = body_stream) do
+      build(method, url, headers, {:stream, body_stream})
     end
 
-    defp request(_name, _method, _url, _headers, %Stream{}, _opts) do
-      raise "Streaming is not supported by this adapter!"
+    defp build(method, url, headers, body_stream_fun) when is_function(body_stream_fun) do
+      build(method, url, headers, {:stream, body_stream_fun})
     end
 
-    defp request(name, method, url, headers, body, opts) do
+    defp build(method, url, headers, body) do
       Finch.build(method, url, headers, body)
-      |> Finch.request(name, opts)
+    end
+
+    defp request(req, name, req_opts, opts) do
+      case opts[:response] do
+        :stream -> stream(req, name, req_opts)
+        nil -> Finch.request(req, name, req_opts)
+        other -> raise "Unknown response option: #{inspect(other)}"
+      end
+    end
+
+    defp stream(req, name, opts) do
+      owner = self()
+      ref = make_ref()
+
+      fun = fn
+        {:status, status}, _acc -> status
+        {:headers, headers}, status -> send(owner, {ref, {:status, status, headers}})
+        {:data, data}, _acc -> send(owner, {ref, {:data, data}})
+      end
+
+      task =
+        Task.async(fn ->
+          case Finch.stream(req, name, nil, fun, opts) do
+            {:ok, _acc} -> send(owner, {ref, :eof})
+            {:error, error} -> send(owner, {ref, {:error, error}})
+          end
+        end)
+
+      receive do
+        {^ref, {:status, status, headers}} ->
+          body =
+            Stream.unfold(nil, fn _ ->
+              receive do
+                {^ref, {:data, data}} ->
+                  {data, nil}
+
+                {^ref, :eof} ->
+                  Task.await(task)
+                  nil
+              after
+                opts[:receive_timeout] -> nil
+              end
+            end)
+
+          {:ok, %Finch.Response{status: status, headers: headers, body: body}}
+      after
+        opts[:receive_timeout] ->
+          {:error, :timeout}
+      end
     end
   end
 end

Diff for: mix.exs

@@ -75,7 +75,7 @@ defmodule Tesla.Mixfile do
       {:excoveralls, "~> 0.8", only: :test},
       {:httparrot, "~> 1.3", only: :test},
       {:ex_doc, "~> 0.21", only: :dev, runtime: false},
-      {:mix_test_watch, "~> 1.0", only: :dev},
+      {:mix_test_watch, "~> 1.0", only: :test, runtime: false},
       {:dialyxir, "~> 1.0", only: [:dev, :test], runtime: false},
       {:inch_ex, "~> 2.0", only: :docs},
 

Diff for: test/support/adapter_case/stream_request_body.ex

@@ -3,7 +3,7 @@ defmodule Tesla.AdapterCase.StreamRequestBody do
     quote do
       alias Tesla.Env
 
-      describe "Stream" do
+      describe "Stream Request" do
         test "stream request body: Stream.map" do
           request = %Env{
             method: :post,

Diff for: test/support/adapter_case/stream_response_body.ex

@@ -0,0 +1,23 @@
+defmodule Tesla.AdapterCase.StreamResponseBody do
+  defmacro __using__(_) do
+    quote do
+      alias Tesla.Env
+
+      describe "Stream Response" do
+        test "stream response body" do
+          request = %Env{
+            method: :get,
+            url: "#{@http}/stream/20"
+          }
+
+          assert {:ok, %Env{} = response} = call(request, response: :stream)
+          assert response.status == 200
+          assert is_function(response.body) || response.body.__struct__ == Stream
+
+          body = Enum.to_list(response.body)
+          assert Enum.count(body) == 20
+        end
+      end
+    end
+  end
+end

Diff for: test/tesla/adapter/finch_test.exs

@@ -6,7 +6,8 @@ defmodule Tesla.Adapter.FinchTest do
   use Tesla.AdapterCase, adapter: {Tesla.Adapter.Finch, [name: @finch_name]}
   use Tesla.AdapterCase.Basic
   use Tesla.AdapterCase.Multipart
-  # use Tesla.AdapterCase.StreamRequestBody
+  use Tesla.AdapterCase.StreamRequestBody
+  use Tesla.AdapterCase.StreamResponseBody
   use Tesla.AdapterCase.SSL
 
   setup do