Add stream_response option to hackney adapter

lib/tesla/adapter/hackney.ex

@@ -42,7 +42,10 @@ if Code.ensure_loaded?(:hackney) do
     end
 
     defp format_body(data) when is_list(data), do: IO.iodata_to_binary(data)
-    defp format_body(data) when is_binary(data) or is_reference(data), do: data
+
+    defp format_body(data)
+         when is_binary(data) or is_reference(data) or is_function(data),
+         do: data
 
     defp request(env, opts) do
       request(
@@ -68,7 +71,12 @@ if Code.ensure_loaded?(:hackney) do
     end
 
     defp request(method, url, headers, body, opts) do
-      handle(:hackney.request(method, url, headers, body || '', opts))
+      response = :hackney.request(method, url, headers, body || '', opts)
+
+      case Keyword.get(opts, :stream_to_pid) do
+        pid when is_pid(pid) -> handle_stream(response, pid)
+        _ -> handle(response)
+      end
     end
 
     defp request_stream(method, url, headers, body, opts) do
@@ -106,6 +114,40 @@ if Code.ensure_loaded?(:hackney) do
 
     defp handle({:ok, status, headers, body}), do: {:ok, status, headers, body}
 
+    defp handle_stream({:ok, status, headers, ref}, pid) when is_reference(ref) do
+      :hackney.controlling_process(ref, pid)
+
+      body =
+        Stream.resource(
+          fn -> nil end,
+          fn _ ->
+            case :hackney.stream_body(ref) do
+              :done ->
+                {:halt, nil}
+
+              {:ok, data} ->
+                {[data], nil}
+
+              {:error, reason} ->
+                raise inspect(reason)
+            end
+          end,
+          fn _ -> :hackney.close(ref) end
+        )
+
+      {:ok, status, headers, body}
+    end
+
+    defp handle_stream(response, pid) do
+      case handle(response) do
+        {:ok, _status, _headers, ref} = response when is_reference(ref) ->
+          handle_stream(response, pid)
+
+        response ->
+          response
+      end
+    end
+
     defp handle_async_response({ref, %{headers: headers, status: status}})
          when not (is_nil(headers) or is_nil(status)) do
       {:ok, status, headers, ref}

test/tesla/adapter/hackney_test.exs

@@ -62,4 +62,17 @@ defmodule Tesla.Adapter.HackneyTest do
 
     assert {:error, :fake_error} = call(request)
   end
+
+  test "get with `stream_to_pid: pid` option" do
+    request = %Env{
+      method: :get,
+      url: "#{@http}/ip"
+    }
+
+    assert {:ok, %Env{} = response} = call(request, stream_to_pid: self())
+
+    assert response.status == 200
+    assert is_function(response.body)
+    assert is_bitstring(Enum.join(response.body))
+  end
 end