Repository: tallarium/reverse_proxy_plug (elixir)

Issue:
Cannot get proxy to work with Tesla adapter
Hello,

I am trying out this plug, and ran into a strange issue. Configuring tesla_client option was also kind of tricky, and I don't know if I have it setup correctly. The problem I see is that when configured with Tesla (using either mint or hackney as tesla_client), the requests just hang and never complete. Using hackney directly instead of tesla seems to work.

Here's how I have it setup:

mix.exs:
```
defp deps do
  [
    {:reverse_proxy_plug, "~> 2.1"},
    {:tesla, "~> 1.4"},
    {:mint, "~> 1.4"},
    {:castore, "~> 0.1"}
  ]
end
```

endpoint.ex:
```
defmodule MiddleMan.Endpoint do
  use Plug.Router

  plug(Plug.Telemetry, event_prefix: [:middle_man, :plug])
  plug(Plug.Logger)
  plug(:match)

  plug(:dispatch)

  get "/hello" do
    send_resp(conn, 200, "world")
  end

forward("/test",
    to: ReverseProxyPlug,
    client: ReverseProxyPlug.HTTPClient.Adapters.Tesla,
    upstream: "//localhost:8000/",
    client_options: [
      tesla_client: Tesla.client([], {Tesla.Adapter.Mint, []})
    ]
  )
end
```

When I make a request to http://localhost:4000/test/test.json, I see this in the logs:
```15:21:59.484 [info]  GET /test/test.json```
and the browser keeps loading and loading with no response.

Any clues what I did wrong above? I was unable to figure out how to pass tesla_client option in config.exs, so I ended up passing the options in the forward call.

Any help would be appreciated.....thanks for your time!

Pull request description:
refactor stream response handling to optional callback
fixes #148 

and throw an error if the client adapter does not support the response mode configured.

Patch:
diff --git a/lib/reverse_proxy_plug.ex b/lib/reverse_proxy_plug.ex
index 6860385..4ea5f2d 100644
--- a/lib/reverse_proxy_plug.ex
+++ b/lib/reverse_proxy_plug.ex
@@ -44,6 +44,7 @@ defmodule ReverseProxyPlug do
       {m, f, a} -> {m, f, a}
       fun when is_function(fun) -> fun
     end)
+    |> ensure_response_mode_compatibility()
   end
 
   @spec call(Plug.Conn.t(), Keyword.t()) :: Plug.Conn.t()
@@ -183,32 +184,26 @@ defmodule ReverseProxyPlug do
     |> Conn.resp(status, body)
   end
 
-  if {:module, HTTPoison} == Code.ensure_loaded(HTTPoison) do
-    # This section of the code is present for retrocompatibility
-    # for the case where HTTPoison is the underlying HTTP Client
-    defp process_response(:stream, conn, _resp, opts),
-      do: stream_response(conn, opts)
-
-    @spec stream_response(Conn.t(), Keyword.t()) :: Conn.t()
-    defp stream_response(conn, opts) do
-      receive do
-        %HTTPoison.AsyncStatus{code: code} ->
-          case opts[:status_callbacks][code] do
-            nil ->
-              conn
-              |> Conn.put_status(code)
-              |> stream_response(opts)
-
-            handler ->
-              handler.(conn, opts)
-          end
-
-        %HTTPoison.AsyncHeaders{headers: headers} ->
-          additional_headers =
-            if conn.status >= 200 and conn.status != 204,
-              do: [{"transfer-encoding", "chunked"}],
-              else: []
+  defp process_response(:stream, initial_conn, %HTTPClient.AsyncResponse{} = resp, opts) do
+    resp
+    |> opts[:client].stream_response()
+    |> Enum.reduce_while(initial_conn, fn
+      {:status, status}, conn ->
+        case opts[:status_callbacks][status] do
+          nil ->
+            {:cont, conn |> Conn.put_status(status)}
 
+          handler ->
+            {:halt, handler.(conn, opts)}
+        end
+
+      {:headers, headers}, conn ->
+        additional_headers =
+          if conn.status >= 200 and conn.status != 204,
+            do: [{"transfer-encoding", "chunked"}],
+            else: []
+
+        conn =
           headers
           |> opts[:normalize_headers].()
           |> remove_hop_by_hop_headers
@@ -216,28 +211,18 @@ defmodule ReverseProxyPlug do
           |> Enum.concat(additional_headers)
           |> add_resp_headers(conn, opts[:stream_headers_mode])
           |> Conn.send_chunked(conn.status)
-          |> stream_response(opts)
-
-        %HTTPoison.AsyncChunk{chunk: chunk} ->
-          case Conn.chunk(conn, chunk) do
-            {:ok, conn} ->
-              stream_response(conn, opts)
-
-            {:error, :closed} ->
-              conn
-          end
-
-        %HTTPoison.AsyncEnd{} ->
-          conn
-
-        %HTTPoison.Error{reason: reason} ->
-          do_error_callback(
-            opts[:error_callback],
-            {:error, %HTTPClient.Error{reason: reason}},
-            conn
-          )
-      end
-    end
+
+        {:cont, conn}
+
+      {:chunk, chunk}, conn ->
+        case Conn.chunk(conn, chunk) do
+          {:ok, conn_after_chunk} -> {:cont, conn_after_chunk}
+          {:error, _error} -> {:halt, conn}
+        end
+
+      {:error, error}, conn ->
+        {:halt, do_error_callback(opts[:error_callback], error, conn)}
+    end)
   end
 
   defp prepare_url(conn, overrides) do
@@ -435,6 +420,16 @@ defmodule ReverseProxyPlug do
     end
   end
 
+  defp ensure_response_mode_compatibility(opts) do
+    if opts[:response_mode] == :stream and
+         not function_exported?(opts[:client], :stream_response, 1) do
+      raise ArgumentError,
+            "The client adapter does not support streaming responses. Please use :buffer response mode."
+    else
+      opts
+    end
+  end
+
   defp add_resp_headers(resp_headers, conn, :prepend) do
     Conn.prepend_resp_headers(conn, resp_headers)
   end
diff --git a/lib/reverse_proxy_plug/http_client.ex b/lib/reverse_proxy_plug/http_client.ex
index 2c1f6ed..5491aa6 100644
--- a/lib/reverse_proxy_plug/http_client.ex
+++ b/lib/reverse_proxy_plug/http_client.ex
@@ -10,4 +10,8 @@ defmodule ReverseProxyPlug.HTTPClient do
                | __MODULE__.AsyncResponse.t()
                | __MODULE__.MaybeRedirect.t()}
               | {:error, error()}
+
+  @callback stream_response(__MODULE__.AsyncResponse.t()) :: Enumerable.t()
+
+  @optional_callbacks stream_response: 1
 end
diff --git a/lib/reverse_proxy_plug/http_client/adapters/httpoison.ex b/lib/reverse_proxy_plug/http_client/adapters/httpoison.ex
index 925e8ea..3f384dc 100644
--- a/lib/reverse_proxy_plug/http_client/adapters/httpoison.ex
+++ b/lib/reverse_proxy_plug/http_client/adapters/httpoison.ex
@@ -68,5 +68,27 @@ if Code.ensure_loaded?(HTTPoison) do
     defp translate_mod(HTTPoison.AsyncResponse), do: HTTPClient.AsyncResponse
     defp translate_mod(HTTPoison.MaybeRedirect), do: HTTPClient.MaybeRedirect
     defp translate_mod(HTTPoison.Error), do: HTTPClient.Error
+
+    @impl HTTPClient
+    def stream_response(_resp) do
+      Stream.unfold(nil, fn _ ->
+        receive do
+          %HTTPoison.AsyncStatus{code: code} ->
+            {{:status, code}, nil}
+
+          %HTTPoison.AsyncHeaders{headers: headers} ->
+            {{:headers, headers}, nil}
+
+          %HTTPoison.AsyncChunk{chunk: chunk} ->
+            {{:chunk, chunk}, nil}
+
+          %HTTPoison.Error{reason: reason} ->
+            {{:error, %HTTPClient.Error{reason: reason}}, nil}
+
+          %HTTPoison.AsyncEnd{} ->
+            nil
+        end
+      end)
+    end
   end
 end


Test patch:
diff --git a/test/reverse_proxy_plug_test.exs b/test/reverse_proxy_plug_test.exs
index 1a84c56..4772443 100644
--- a/test/reverse_proxy_plug_test.exs
+++ b/test/reverse_proxy_plug_test.exs
@@ -83,6 +83,20 @@ defmodule ReverseProxyPlugTest do
     assert conn.resp_body == Jason.encode!(body)
   end
 
+  test "throws an error if :stream mode is used with the Tesla adapter" do
+    proxy_opts =
+      Keyword.merge(@opts,
+        client: ReverseProxyPlug.HTTPClient.Adapters.Tesla,
+        client_options: [tesla_client: Tesla.client([], ReverseProxyPlug.TeslaMock)],
+        response_mode: :stream
+      )
+
+    assert_raise ArgumentError, fn ->
+      conn(:get, "/")
+      |> ReverseProxyPlug.call(ReverseProxyPlug.init(proxy_opts))
+    end
+  end
+
   test "does not add transfer-encoding header to response" do
     headers = [{"host", "example.com"}, {"content-length", "42"}]
 
@@ -126,6 +140,7 @@ defmodule ReverseProxyPlugTest do
   test "receives stream response" do
     ReverseProxyPlug.HTTPClientMock
     |> expect(:request, TestReuse.get_stream_responder(%{headers: @host_header}))
+    |> stub(:stream_response, &HTTPClient.Adapters.HTTPoison.stream_response/1)
 
     conn =
       conn(:get, "/")
@@ -139,6 +154,7 @@ defmodule ReverseProxyPlugTest do
   test "sets correct chunked transfer-encoding headers" do
     ReverseProxyPlug.HTTPClientMock
     |> expect(:request, TestReuse.get_stream_responder(%{headers: [{"content-length", "7"}]}))
+    |> stub(:stream_response, &HTTPClient.Adapters.HTTPoison.stream_response/1)
 
     conn =
       conn(:get, "/")
@@ -158,6 +174,7 @@ defmodule ReverseProxyPlugTest do
   test "does not sets transfer-encoding headers for informational status class" do
     ReverseProxyPlug.HTTPClientMock
     |> expect(:request, TestReuse.get_stream_responder(%{status_code: 100}))
+    |> stub(:stream_response, &HTTPClient.Adapters.HTTPoison.stream_response/1)
 
     conn =
       conn(:get, "/")
@@ -174,6 +191,7 @@ defmodule ReverseProxyPlugTest do
   test "does not sets transfer-encoding headers for no content status code" do
     ReverseProxyPlug.HTTPClientMock
     |> expect(:request, TestReuse.get_stream_responder(%{status_code: 204}))
+    |> stub(:stream_response, &HTTPClient.Adapters.HTTPoison.stream_response/1)
 
     conn =
       conn(:post, "/", nil)
@@ -230,6 +248,7 @@ defmodule ReverseProxyPlugTest do
   test "calls status callback" do
     ReverseProxyPlug.HTTPClientMock
     |> expect(:request, TestReuse.get_stream_responder(%{status_code: 500}))
+    |> stub(:stream_response, &HTTPClient.Adapters.HTTPoison.stream_response/1)
 
     opts =
       @opts
@@ -735,6 +754,7 @@ defmodule ReverseProxyPlugTest do
              ReverseProxyPlug.init(
                upstream: "",
                error_callback: {__MODULE__, :error_handler, []},
+               response_mode: :buffer,
                client: adapter
              )
 
@@ -750,6 +770,7 @@ defmodule ReverseProxyPlugTest do
              ReverseProxyPlug.init(
                upstream: "",
                error_callback: {__MODULE__, :error_handler, []},
+               response_mode: :buffer,
                client: nil
              )
 
diff --git a/test/support/test_reuse.ex b/test/support/test_reuse.ex
index c182339..39bbfa5 100644
--- a/test/support/test_reuse.ex
+++ b/test/support/test_reuse.ex
@@ -3,6 +3,7 @@ defmodule TestReuse do
   @default_opts upstream: "example.com", client: ReverseProxyPlug.HTTPClientMock
 
   alias ReverseProxyPlug.HTTPClient
+  alias ReverseProxyPlug.HTTPClient.Adapters.HTTPoison, as: HTTPoisonAdapter
 
   def get_buffer_responder(response_args) do
     fn request ->
@@ -38,6 +39,9 @@ defmodule TestReuse do
           get_responder: &TestReuse.get_stream_responder/1
         }
 
+        ReverseProxyPlug.HTTPClientMock
+        |> stub(:stream_response, &HTTPoisonAdapter.stream_response/1)
+
         unquote(body)
       end