Repository: tallarium/reverse_proxy_plug
Pull request #210 (2024-04-30 19:15:09)

refactor to separate interface for requests with streamed responses
so we can support other adapters which require special options to be provided to the `request` function


Patch:
diff --git a/lib/reverse_proxy_plug.ex b/lib/reverse_proxy_plug.ex
index 72ede92..fd65253 100644
--- a/lib/reverse_proxy_plug.ex
+++ b/lib/reverse_proxy_plug.ex
@@ -104,15 +104,13 @@ defmodule ReverseProxyPlug do
   end
 
   def request(conn, body, opts) do
-    {method, url, headers, client_options} = prepare_request(conn, opts)
+    request = prepare_request(conn, opts, body)
 
-    opts[:client].request(%HTTPClient.Request{
-      method: method,
-      url: url,
-      body: body,
-      headers: headers,
-      options: client_options
-    })
+    if opts[:response_mode] == :stream do
+      opts[:client].request_stream(request)
+    else
+      opts[:client].request(request)
+    end
   end
 
   def response({:ok, resp}, conn, opts) do
@@ -184,9 +182,8 @@ defmodule ReverseProxyPlug do
     |> Conn.resp(status, body)
   end
 
-  defp process_response(:stream, initial_conn, %HTTPClient.AsyncResponse{} = resp, opts) do
+  defp process_response(:stream, initial_conn, resp, opts) do
     resp
-    |> opts[:client].stream_response()
     |> Enum.reduce_while(initial_conn, fn
       {:status, status}, conn ->
         case Map.fetch(opts[:status_callbacks], status) do
@@ -237,7 +234,7 @@ defmodule ReverseProxyPlug do
     end
   end
 
-  defp prepare_request(conn, options) do
+  defp prepare_request(conn, options, body) do
     method =
       try do
         conn.method
@@ -268,24 +265,18 @@ defmodule ReverseProxyPlug do
     headers = List.keystore(headers, "host", 0, {"host", proxy_req_host})
 
     client_options =
-      options[:response_mode]
-      |> get_client_opts(options[:client_options])
+      options[:client_options]
+      |> Keyword.put_new(:timeout, :infinity)
+      |> Keyword.put_new(:recv_timeout, :infinity)
       |> recycle_cookies(conn)
 
-    {method, url, headers, client_options}
-  end
-
-  defp get_client_opts(:stream, opts) do
-    opts
-    |> Keyword.put_new(:timeout, :infinity)
-    |> Keyword.put_new(:recv_timeout, :infinity)
-    |> Keyword.put_new(:stream_to, self())
-  end
-
-  defp get_client_opts(:buffer, opts) do
-    opts
-    |> Keyword.put_new(:timeout, :infinity)
-    |> Keyword.put_new(:recv_timeout, :infinity)
+    %HTTPClient.Request{
+      method: method,
+      url: url,
+      body: body,
+      headers: headers,
+      options: client_options
+    }
   end
 
   defp send_stream_response_headers(%{status: status} = conn, headers, opts) do
@@ -409,7 +400,7 @@ defmodule ReverseProxyPlug do
 
   defp ensure_response_mode_compatibility(opts) do
     if opts[:response_mode] == :stream and
-         not function_exported?(opts[:client], :stream_response, 1) do
+         not function_exported?(opts[:client], :request_stream, 1) do
       raise ArgumentError,
             "The client adapter does not support streaming responses. Please use :buffer response mode."
     else
diff --git a/lib/reverse_proxy_plug/http_client.ex b/lib/reverse_proxy_plug/http_client.ex
index 5491aa6..fc1bfb7 100644
--- a/lib/reverse_proxy_plug/http_client.ex
+++ b/lib/reverse_proxy_plug/http_client.ex
@@ -7,11 +7,10 @@ defmodule ReverseProxyPlug.HTTPClient do
   @callback request(__MODULE__.Request.t()) ::
               {:ok,
                __MODULE__.Response.t()
-               | __MODULE__.AsyncResponse.t()
                | __MODULE__.MaybeRedirect.t()}
               | {:error, error()}
 
-  @callback stream_response(__MODULE__.AsyncResponse.t()) :: Enumerable.t()
+  @callback request_stream(__MODULE__.Request.t()) :: {:ok, Enumerable.t()} | {:error, error()}
 
-  @optional_callbacks stream_response: 1
+  @optional_callbacks request_stream: 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 3f384dc..069b062 100644
--- a/lib/reverse_proxy_plug/http_client/adapters/httpoison.ex
+++ b/lib/reverse_proxy_plug/http_client/adapters/httpoison.ex
@@ -64,31 +64,38 @@ if Code.ensure_loaded?(HTTPoison) do
       {tag, translated_resp}
     end
 
+    defp translate_mod(HTTPoison.AsyncResponse), do: HTTPClient.Response
     defp translate_mod(HTTPoison.Response), do: HTTPClient.Response
-    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)
+    def request_stream(%HTTPClient.Request{options: options} = req) do
+      case request(%{req | options: Keyword.put(options, :stream_to, self())}) do
+        {:ok, _} ->
+          {:ok,
+           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)}
+
+        {:error, _} = error ->
+          error
+      end
     end
   end
 end
diff --git a/lib/reverse_proxy_plug/http_client/async_response.ex b/lib/reverse_proxy_plug/http_client/async_response.ex
deleted file mode 100644
index 397bf81..0000000
--- a/lib/reverse_proxy_plug/http_client/async_response.ex
+++ /dev/null
@@ -1,5 +0,0 @@
-defmodule ReverseProxyPlug.HTTPClient.AsyncResponse do
-  @moduledoc "Carries the reference to be used for parsing an asynchronous response"
-  defstruct id: nil
-  @type t :: %__MODULE__{id: reference | nil}
-end


Test patch:
diff --git a/test/http_client/adapters/httpoison_test.exs b/test/http_client/adapters/httpoison_test.exs
index eeb5834..7112fae 100644
--- a/test/http_client/adapters/httpoison_test.exs
+++ b/test/http_client/adapters/httpoison_test.exs
@@ -4,7 +4,6 @@ defmodule ReverseProxyPlug.HTTPClient.Adapters.HTTPoisonTest do
   alias ReverseProxyPlug.HTTPClient.Adapters.HTTPoison, as: HTTPoisonClient
 
   alias ReverseProxyPlug.HTTPClient.{
-    AsyncResponse,
     Error,
     Request,
     Response
@@ -40,8 +39,7 @@ defmodule ReverseProxyPlug.HTTPClient.Adapters.HTTPoisonTest do
 
         req = %Request{
           method: unquote(method),
-          url: "http://localhost:8000#{path}",
-          options: [stream_to: self()]
+          url: "http://localhost:8000#{path}"
         }
 
         Bypass.expect_once(bypass, fn %Plug.Conn{} = conn ->
@@ -50,11 +48,13 @@ defmodule ReverseProxyPlug.HTTPClient.Adapters.HTTPoisonTest do
           Plug.Conn.send_resp(conn, 204, "")
         end)
 
-        assert {:ok, %AsyncResponse{id: id}} = HTTPoisonClient.request(req)
+        assert {:ok, stream} = HTTPoisonClient.request_stream(req)
+
+        assert [
+                 {:status, 204},
+                 {:headers, headers}
+               ] = Enum.to_list(stream)
 
-        assert_receive %HTTPoison.AsyncStatus{id: ^id, code: 204}, 1_000
-        assert_receive %HTTPoison.AsyncHeaders{id: ^id, headers: headers}, 1_000
-        assert_receive %HTTPoison.AsyncEnd{id: ^id}, 1_000
         assert is_list(headers)
       end
 
@@ -94,9 +94,9 @@ defmodule ReverseProxyPlug.HTTPClient.Adapters.HTTPoisonTest do
         Plug.Conn.send_resp(conn, 204, "")
       end)
 
-      assert {:ok, %AsyncResponse{id: id}} = HTTPoisonClient.request(req)
+      assert {:ok, %Response{}} = HTTPoisonClient.request(req)
 
-      assert_receive %HTTPoison.Error{id: ^id, reason: {:closed, :timeout}}, 1_000
+      assert_receive %HTTPoison.Error{reason: {:closed, :timeout}}, 1_000
     end
   end
 end
diff --git a/test/reverse_proxy_plug_test.exs b/test/reverse_proxy_plug_test.exs
index 5e231b6..9c6fc19 100644
--- a/test/reverse_proxy_plug_test.exs
+++ b/test/reverse_proxy_plug_test.exs
@@ -143,8 +143,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)
+    |> expect(:request_stream, TestReuse.get_stream_responder(%{headers: @host_header}))
 
     conn =
       conn(:get, "/")
@@ -197,8 +196,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)
+    |> expect(:request_stream, TestReuse.get_stream_responder(%{status_code: 500}))
 
     opts =
       @opts
@@ -250,10 +248,10 @@ defmodule ReverseProxyPlugTest do
   end
 
   test_stream_and_buffer "adds x-forwarded-for header if not present" do
-    %{opts: opts, get_responder: get_responder} = test_reuse_opts
+    %{req_function: req_function, opts: opts, get_responder: get_responder} = test_reuse_opts
 
     ReverseProxyPlug.HTTPClientMock
-    |> expect(:request, fn %{headers: headers} = request ->
+    |> expect(req_function, fn %{headers: headers} = request ->
       send(self(), {:headers, headers})
       get_responder.(%{}).(request)
     end)
@@ -266,10 +264,10 @@ defmodule ReverseProxyPlugTest do
   end
 
   test_stream_and_buffer "appends to x-forwarded-for header if present" do
-    %{opts: opts, get_responder: get_responder} = test_reuse_opts
+    %{req_function: req_function, opts: opts, get_responder: get_responder} = test_reuse_opts
 
     ReverseProxyPlug.HTTPClientMock
-    |> expect(:request, fn %{headers: headers} = request ->
+    |> expect(req_function, fn %{headers: headers} = request ->
       send(self(), {:headers, headers})
       get_responder.(%{}).(request)
     end)
@@ -283,10 +281,10 @@ defmodule ReverseProxyPlugTest do
   end
 
   test_stream_and_buffer "removes hop-by-hop headers before forwarding request" do
-    %{opts: opts, get_responder: get_responder} = test_reuse_opts
+    %{req_function: req_function, opts: opts, get_responder: get_responder} = test_reuse_opts
 
     ReverseProxyPlug.HTTPClientMock
-    |> expect(:request, fn %{headers: headers} = request ->
+    |> expect(req_function, fn %{headers: headers} = request ->
       send(self(), {:headers, headers})
       get_responder.(%{}).(request)
     end)
@@ -300,11 +298,11 @@ defmodule ReverseProxyPlugTest do
   end
 
   test_stream_and_buffer "removes hop-by-hop headers from response" do
-    %{opts: opts, get_responder: get_responder} = test_reuse_opts
+    %{req_function: req_function, opts: opts, get_responder: get_responder} = test_reuse_opts
 
     ReverseProxyPlug.HTTPClientMock
     |> expect(
-      :request,
+      req_function,
       get_responder.(%{headers: @hop_by_hop_headers ++ @end_to_end_headers})
     )
 
@@ -320,10 +318,10 @@ defmodule ReverseProxyPlugTest do
   end
 
   test_stream_and_buffer "returns bad gateway on error" do
-    %{opts: opts} = test_reuse_opts
+    %{req_function: req_function, opts: opts} = test_reuse_opts
 
     ReverseProxyPlug.HTTPClientMock
-    |> expect(:request, fn _request ->
+    |> expect(req_function, fn _request ->
       {:error, %HTTPClient.Error{reason: :no_reason}}
     end)
 
@@ -359,12 +357,12 @@ defmodule ReverseProxyPlugTest do
   end
 
   test_stream_and_buffer "calls error callback if supplied" do
-    %{opts: opts} = test_reuse_opts
+    %{req_function: req_function, opts: opts} = test_reuse_opts
 
     error = %HTTPClient.Error{reason: :no_reason}
 
     ReverseProxyPlug.HTTPClientMock
-    |> expect(:request, fn _request ->
+    |> expect(req_function, fn _request ->
       {:error, error}
     end)
 
@@ -378,10 +376,10 @@ defmodule ReverseProxyPlugTest do
   end
 
   test_stream_and_buffer "calls error callback with arity 2 if supplied" do
-    %{opts: opts} = test_reuse_opts
+    %{req_function: req_function, opts: opts} = test_reuse_opts
 
     ReverseProxyPlug.HTTPClientMock
-    |> expect(:request, fn _request ->
+    |> expect(req_function, fn _request ->
       {:error, @error}
     end)
 
@@ -397,10 +395,10 @@ defmodule ReverseProxyPlugTest do
   end
 
   test_stream_and_buffer "calls error callback if supplied as MFA tuple" do
-    %{opts: opts} = test_reuse_opts
+    %{req_function: req_function, opts: opts} = test_reuse_opts
 
     ReverseProxyPlug.HTTPClientMock
-    |> expect(:request, fn _request ->
+    |> expect(req_function, fn _request ->
       {:error, @error}
     end)
 
@@ -415,10 +413,10 @@ defmodule ReverseProxyPlugTest do
   end
 
   test_stream_and_buffer "calls error callback with higher arity if present" do
-    %{opts: opts} = test_reuse_opts
+    %{req_function: req_function, opts: opts} = test_reuse_opts
 
     ReverseProxyPlug.HTTPClientMock
-    |> expect(:request, fn _request ->
+    |> expect(req_function, fn _request ->
       {:error, @error}
     end)
 
@@ -436,12 +434,12 @@ defmodule ReverseProxyPlugTest do
   end
 
   test_stream_and_buffer "handles request path and query string" do
-    %{opts: opts, get_responder: get_responder} = test_reuse_opts
+    %{req_function: req_function, opts: opts, get_responder: get_responder} = test_reuse_opts
 
     opts_with_upstream = Keyword.merge(opts, upstream: "//example.com/root_upstream?query=yes")
 
     ReverseProxyPlug.HTTPClientMock
-    |> expect(:request, fn %{url: url} = request ->
+    |> expect(req_function, fn %{url: url} = request ->
       send(self(), {:url, url})
       get_responder.(%{}).(request)
     end)
@@ -454,11 +452,11 @@ defmodule ReverseProxyPlugTest do
   end
 
   test_stream_and_buffer "preserves trailing slash at the end of request path" do
-    %{opts: opts, get_responder: get_responder} = test_reuse_opts
+    %{req_function: req_function, opts: opts, get_responder: get_responder} = test_reuse_opts
     opts_with_upstream = Keyword.merge(opts, upstream: "//example.com")
 
     ReverseProxyPlug.HTTPClientMock
-    |> expect(:request, fn %{url: url} = request ->
+    |> expect(req_function, fn %{url: url} = request ->
       send(self(), {:url, url})
       get_responder.(%{}).(request)
     end)
@@ -471,11 +469,11 @@ defmodule ReverseProxyPlugTest do
   end
 
   test_stream_and_buffer "don't add a redundant slash at the end of request path" do
-    %{opts: opts, get_responder: get_responder} = test_reuse_opts
+    %{req_function: req_function, opts: opts, get_responder: get_responder} = test_reuse_opts
     opts_with_upstream = Keyword.merge(opts, upstream: "//example.com")
 
     ReverseProxyPlug.HTTPClientMock
-    |> expect(:request, fn %{url: url} = request ->
+    |> expect(req_function, fn %{url: url} = request ->
       send(self(), {:url, url})
       get_responder.(%{}).(request)
     end)
@@ -488,13 +486,13 @@ defmodule ReverseProxyPlugTest do
   end
 
   test_stream_and_buffer "allow upstream configured at runtime" do
-    %{opts: opts, get_responder: get_responder} = test_reuse_opts
+    %{req_function: req_function, opts: opts, get_responder: get_responder} = test_reuse_opts
 
     opts_with_upstream =
       Keyword.merge(opts, upstream: fn -> "//runtime.com/root_upstream?query=yes" end)
 
     ReverseProxyPlug.HTTPClientMock
-    |> expect(:request, fn %HTTPClient.Request{url: url} = request ->
+    |> expect(req_function, fn %HTTPClient.Request{url: url} = request ->
       send(self(), {:url, url})
       get_responder.(%{}).(request)
     end)
@@ -507,7 +505,7 @@ defmodule ReverseProxyPlugTest do
   end
 
   test_stream_and_buffer "allow upstream configured at runtime with conn data" do
-    %{opts: opts, get_responder: get_responder} = test_reuse_opts
+    %{req_function: req_function, opts: opts, get_responder: get_responder} = test_reuse_opts
 
     opts_with_upstream =
       Keyword.merge(opts,
@@ -518,7 +516,7 @@ defmodule ReverseProxyPlugTest do
       )
 
     ReverseProxyPlug.HTTPClientMock
-    |> expect(:request, 2, fn %HTTPClient.Request{url: url} = request ->
+    |> expect(req_function, 2, fn %HTTPClient.Request{url: url} = request ->
       send(self(), {:url, url})
       get_responder.(%{}).(request)
     end)
@@ -537,11 +535,11 @@ defmodule ReverseProxyPlugTest do
   end
 
   test_stream_and_buffer "include the port in the host header when is not the default and preserve_host_header is false in opts" do
-    %{opts: opts, get_responder: get_responder} = test_reuse_opts
+    %{req_function: req_function, opts: opts, get_responder: get_responder} = test_reuse_opts
     opts_with_upstream = Keyword.merge(opts, upstream: "//example-custom-port.com:8081")
 
     ReverseProxyPlug.HTTPClientMock
-    |> expect(:request, fn %{headers: headers} = request ->
+    |> expect(req_function, fn %{headers: headers} = request ->
       send(self(), {:headers, headers})
       get_responder.(%{}).(request)
     end)
@@ -554,11 +552,11 @@ defmodule ReverseProxyPlugTest do
   end
 
   test_stream_and_buffer "don't include the port in the host header when is the default and preserve_host_header is false in opts" do
-    %{opts: opts, get_responder: get_responder} = test_reuse_opts
+    %{req_function: req_function, opts: opts, get_responder: get_responder} = test_reuse_opts
     opts_with_upstream = Keyword.merge(opts, upstream: "//example-custom-port.com:80")
 
     ReverseProxyPlug.HTTPClientMock
-    |> expect(:request, fn %{headers: headers} = request ->
+    |> expect(req_function, fn %{headers: headers} = request ->
       send(self(), {:headers, headers})
       get_responder.(%{}).(request)
     end)
@@ -571,11 +569,11 @@ defmodule ReverseProxyPlugTest do
   end
 
   test_stream_and_buffer "don't include the port in the host header when is the default for https and preserve_host_header is false in opts" do
-    %{opts: opts, get_responder: get_responder} = test_reuse_opts
+    %{req_function: req_function, opts: opts, get_responder: get_responder} = test_reuse_opts
     opts_with_upstream = Keyword.merge(opts, upstream: "https://example-custom-port.com:443")
 
     ReverseProxyPlug.HTTPClientMock
-    |> expect(:request, fn %{headers: headers} = request ->
+    |> expect(req_function, fn %{headers: headers} = request ->
       send(self(), {:headers, headers})
       get_responder.(%{}).(request)
     end)
@@ -589,31 +587,32 @@ defmodule ReverseProxyPlugTest do
 
   for timeout_reason <- ReverseProxyPlug.get_timeout_error_reasons() do
     test_stream_and_buffer "returns gateway timeout on #{inspect(timeout_reason)} as error reason" do
-      %{opts: opts} = test_reuse_opts
+      %{req_function: req_function, opts: opts} = test_reuse_opts
 
-      conn = :get |> conn("/") |> simulate_upstream_error(unquote(timeout_reason), opts)
+      conn =
+        :get |> conn("/") |> simulate_upstream_error(unquote(timeout_reason), opts, req_function)
 
       assert conn.status === 504
     end
   end
 
   test_stream_and_buffer "returns gateway error on a generic error" do
-    %{opts: opts} = test_reuse_opts
+    %{req_function: req_function, opts: opts} = test_reuse_opts
 
-    conn = :get |> conn("/") |> simulate_upstream_error(:some_error, opts)
+    conn = :get |> conn("/") |> simulate_upstream_error(:some_error, opts, req_function)
 
     assert conn.status === 502
   end
 
   test_stream_and_buffer "passes timeout options to HTTP client" do
-    %{opts: opts, get_responder: get_responder} = test_reuse_opts
+    %{req_function: req_function, opts: opts, get_responder: get_responder} = test_reuse_opts
     timeout_val = 5_000
 
     opts_with_client_options =
       Keyword.merge(opts, client_options: [timeout: timeout_val, recv_timeout: timeout_val])
 
     ReverseProxyPlug.HTTPClientMock
-    |> expect(:request, fn %{options: options} = request ->
+    |> expect(req_function, fn %{options: options} = request ->
       send(self(), {:httpclient_options, options})
       get_responder.(%{}).(request)
     end)
@@ -627,14 +626,14 @@ defmodule ReverseProxyPlugTest do
 
   test_stream_and_buffer "add upstream headers with prepend mode option" do
     upstream_headers = [{"access-control-allow-origin", "*"}]
-    %{opts: opts, get_responder: get_responder} = test_reuse_opts
+    %{req_function: req_function, opts: opts, get_responder: get_responder} = test_reuse_opts
 
     opts_with_headers_mode_prepend =
       Keyword.merge(opts, stream_headers_mode: :prepend, buffer_headers_mode: :prepend)
 
     ReverseProxyPlug.HTTPClientMock
     |> expect(
-      :request,
+      req_function,
       get_responder.(%{headers: upstream_headers})
     )
 
@@ -648,14 +647,14 @@ defmodule ReverseProxyPlugTest do
 
   test_stream_and_buffer "add upstream headers with replace mode option" do
     upstream_headers = [{"access-control-allow-origin", "*"}]
-    %{opts: opts, get_responder: get_responder} = test_reuse_opts
+    %{req_function: req_function, opts: opts, get_responder: get_responder} = test_reuse_opts
 
     opts_with_headers_mode_replace =
       Keyword.merge(opts, stream_headers_mode: :replace, buffer_headers_mode: :replace)
 
     ReverseProxyPlug.HTTPClientMock
     |> expect(
-      :request,
+      req_function,
       get_responder.(%{headers: upstream_headers})
     )
 
@@ -744,10 +743,10 @@ defmodule ReverseProxyPlugTest do
   end
 
   test_stream_and_buffer "recycles cookies from connection" do
-    %{opts: opts, get_responder: get_responder} = test_reuse_opts
+    %{req_function: req_function, opts: opts, get_responder: get_responder} = test_reuse_opts
 
     ReverseProxyPlug.HTTPClientMock
-    |> expect(:request, fn %{options: options} = request ->
+    |> expect(req_function, fn %{options: options} = request ->
       send(self(), {:httpclient_options, options})
       get_responder.(%{}).(request)
     end)
@@ -762,10 +761,10 @@ defmodule ReverseProxyPlugTest do
   end
 
   test_stream_and_buffer "client options do not set cookies if not present on connection" do
-    %{opts: opts, get_responder: get_responder} = test_reuse_opts
+    %{req_function: req_function, opts: opts, get_responder: get_responder} = test_reuse_opts
 
     ReverseProxyPlug.HTTPClientMock
-    |> expect(:request, fn %{options: options} = request ->
+    |> expect(req_function, fn %{options: options} = request ->
       send(self(), {:httpclient_options, options})
       get_responder.(%{}).(request)
     end)
@@ -777,9 +776,9 @@ defmodule ReverseProxyPlugTest do
     refute httpclient_options[:hackney][:cookie]
   end
 
-  defp simulate_upstream_error(conn, reason, opts) do
+  defp simulate_upstream_error(conn, reason, opts, req_function) do
     ReverseProxyPlug.HTTPClientMock
-    |> expect(:request, fn _request ->
+    |> expect(req_function, fn _request ->
       {:error, %HTTPClient.Error{reason: reason}}
     end)
 
diff --git a/test/support/test_reuse.ex b/test/support/test_reuse.ex
index 39bbfa5..9422e3e 100644
--- a/test/support/test_reuse.ex
+++ b/test/support/test_reuse.ex
@@ -3,7 +3,6 @@ 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 ->
@@ -15,19 +14,18 @@ defmodule TestReuse do
     fn request ->
       %{status_code: code, headers: headers, body: body} = make_response(response_args, request)
 
-      send(self(), %HTTPoison.AsyncStatus{code: code})
-      send(self(), %HTTPoison.AsyncHeaders{headers: headers})
+      body_chunks =
+        body
+        |> String.codepoints()
+        |> Enum.chunk_every(body |> String.length() |> div(3))
+        |> Enum.map(&Enum.join/1)
+        |> Enum.map(&{:chunk, &1})
 
-      body
-      |> String.codepoints()
-      |> Enum.chunk_every(body |> String.length() |> div(3))
-      |> Enum.map(&Enum.join/1)
-      |> Enum.each(fn chunk ->
-        send(self(), %HTTPoison.AsyncChunk{chunk: chunk})
-      end)
-
-      send(self(), %HTTPoison.AsyncEnd{})
-      {:ok, %HTTPClient.AsyncResponse{}}
+      {:ok,
+       [
+         {:status, code},
+         {:headers, headers}
+       ] ++ body_chunks}
     end
   end
 
@@ -36,19 +34,18 @@ defmodule TestReuse do
       test unquote(message) <> " (stream)" do
         var!(test_reuse_opts) = %{
           opts: [response_mode: :stream] ++ unquote(@default_opts),
-          get_responder: &TestReuse.get_stream_responder/1
+          get_responder: &TestReuse.get_stream_responder/1,
+          req_function: :request_stream
         }
 
-        ReverseProxyPlug.HTTPClientMock
-        |> stub(:stream_response, &HTTPoisonAdapter.stream_response/1)
-
         unquote(body)
       end
 
       test unquote(message) <> " (buffer)" do
         var!(test_reuse_opts) = %{
           opts: [response_mode: :buffer] ++ unquote(@default_opts),
-          get_responder: &TestReuse.get_buffer_responder/1
+          get_responder: &TestReuse.get_buffer_responder/1,
+          req_function: :request
         }
 
         unquote(body)