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,14 +71,22 @@ 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_response, false) do
+        true -> handle(response, :stream)
+        false -> handle(response)
+      end
     end
 
     defp request_stream(method, url, headers, body, opts) do
       with {:ok, ref} <- :hackney.request(method, url, headers, :stream, opts) do
         case send_stream(ref, body) do
-          :ok -> handle(:hackney.start_response(ref))
-          error -> handle(error)
+          :ok ->
+            handle(:hackney.start_response(ref))
+
+          error ->
+            handle(error)
         end
       else
         e -> handle(e)
@@ -106,6 +117,31 @@ if Code.ensure_loaded?(:hackney) do
 
     defp handle({:ok, status, headers, body}), do: {:ok, status, headers, body}
 
+    defp handle({:ok, status, headers, ref}, :stream) when is_reference(ref) do
+      state = :hackney_manager.get_state(ref)
+
+      body =
+        Stream.resource(
+          fn -> state end,
+          fn
+            {:done, state} ->
+              {:halt, state}
+
+            {:ok, data, state} ->
+              {[data], state}
+
+            {:error, reason} ->
+              raise inspect(reason)
+
+            state ->
+              {[], :hackney_response.stream_body(state)}
+          end,
+          &:hackney_response.close/1
+        )
+
+      {:ok, status, headers, body}
+    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,16 @@ defmodule Tesla.Adapter.HackneyTest do
 
     assert {:error, :fake_error} = call(request)
   end
+
+  test "get with `stream_response: true` option" do
+    request = %Env{
+      method: :get,
+      url: "#{@http}/ip"
+    }
+
+    assert {:ok, %Env{} = response} = call(request, stream_response: true)
+
+    assert response.status == 200
+    assert(is_function(response.body))
+  end
 end