Docker registry proxy support stream response.

Signed-off-by: Edmondfrank <[email protected]>

lib/compass_admin_web/controllers/debug_controller.ex

@@ -1,5 +1,6 @@
 defmodule CompassAdminWeb.DebugController do
   alias Plug.Conn
+  alias ReverseProxyPlug.HTTPClient
   alias CompassAdmin.DockerTokenCacher
   alias CompassAdmin.DockerTokenCacher.Token
 
@@ -23,14 +24,25 @@ defmodule CompassAdminWeb.DebugController do
   end
 
   def docker_registry_proxy(conn, _params) do
+
+    upstream = "https://registry-1.docker.io"
+
     params =
       ReverseProxyPlug.init(
-        upstream: "https://registry-1.docker.io",
+        upstream: upstream,
         client_options: @client_options,
         response_mode: :buffer,
         preserve_host_header: false
       )
 
+    stream_params =
+      ReverseProxyPlug.init(
+        upstream: upstream,
+        client_options: @client_options,
+        response_mode: :stream,
+        preserve_host_header: false
+      )
+
     {:ok, body, conn} = Plug.Conn.read_body(conn)
 
     case DockerTokenCacher.get(conn) do
@@ -43,23 +55,47 @@ defmodule CompassAdminWeb.DebugController do
         auth_conn
         |> ReverseProxyPlug.request(body, params)
         |> handle_redirect(auth_conn)
-        |> ReverseProxyPlug.response(conn, params)
+        |> handle_redirect_resp()
+        |> ReverseProxyPlug.response(conn, if(Enum.member?(conn.path_info, "blobs"), do: stream_params, else: params))
       {:error, reason} ->
         json(conn, %{error: reason})
     end
   end
 
-  def handle_redirect({:ok, %{status_code: code, headers: headers, request: %{headers: req_headers}}}, conn)  when code > 300 and code < 400 do
+  def handle_redirect({:ok, %{status_code: code, headers: headers, request: %{headers: req_headers}}}, conn)
+  when code > 300 and code < 400 do
     next = get_location(headers)
-    method =
-      conn.method
-      |> String.downcase()
-      |> String.to_existing_atom()
-    apply(HTTPoison, method, [next, remove_host(req_headers), @client_options])
+    final_options = if Enum.member?(conn.path_info, "blobs"),
+      do: @client_options |> Keyword.put_new(:stream_to, self()),
+      else: @client_options
+  method =
+    conn.method
+    |> String.downcase()
+    |> String.to_existing_atom()
+  apply(HTTPoison, method, [next, remove_host(req_headers), final_options])
   end
 
   def handle_redirect(resp, _), do: resp
 
+  def handle_redirect_resp({:ok, %HTTPoison.AsyncResponse{} = resp}) do
+    {:ok, translate_response(resp)}
+  end
+
+  def handle_redirect_resp(resp), do: resp
+
+  defp translate_response(%mod{} = response) do
+    data = Map.from_struct(response)
+
+    translated_resp =
+      mod
+      |> translate_mod()
+      |> struct(data)
+
+    translated_resp
+  end
+
+  defp translate_mod(HTTPoison.AsyncResponse), do: HTTPClient.AsyncResponse
+
   defp get_location(headers) do
     {_h, location} =
       Enum.find(headers, fn {header, _location} ->

mix.exs

@@ -73,7 +73,7 @@ defmodule CompassAdmin.MixProject do
       {:jason, "~> 1.2"},
       {:sitemapper, "~> 0.6"},
       {:plug_cowboy, "~> 2.5"},
-      {:reverse_proxy_plug, "~> 3.0"},
+      {:reverse_proxy_plug, "~> 2.4"},
       {:httpoison, "~> 2.2"},
       {:petal_components, "~> 0.18.0"},
       {:ex_indexea, "~> 0.1.0"},