221 - added forwarding of extra headers for response_from

Author: mmacai

lib/rig_api/v3/responses.ex

@@ -37,12 +37,12 @@ defmodule RigApi.V3.Responses do
   """
   def create(%{req_headers: req_headers} = conn, message) do
     case ResponseFromParser.parse(req_headers, message) do
-      {deserialized_pid, response_code, response} ->
+      {deserialized_pid, response_code, response, extra_headers} ->
         Logger.debug(fn ->
           "HTTP response via internal HTTP to #{inspect(deserialized_pid)}: #{inspect(message)}"
         end)
 
-        send(deserialized_pid, {:response_received, response, response_code})
+        send(deserialized_pid, {:response_received, response, response_code, extra_headers})
         send_resp(conn, :accepted, "message sent to correlated reverse proxy request")
 
       err ->

lib/rig_inbound_gateway/api_proxy/handler/http.ex

@@ -103,7 +103,7 @@ defmodule RigInboundGateway.ApiProxy.Handler.Http do
       end
 
     receive do
-      {:response_received, response, response_code} ->
+      {:response_received, response, response_code, extra_headers} ->
         ProxyMetrics.count_proxy_request(
           conn.method,
           conn.request_path,
@@ -115,6 +115,9 @@ defmodule RigInboundGateway.ApiProxy.Handler.Http do
         conn
         |> with_cors()
         |> Tracing.Plug.put_resp_header(Tracing.context())
+        |> Map.update!(:resp_headers, fn existing_headers ->
+          existing_headers ++ Map.to_list(extra_headers)
+        end)
         |> Conn.put_resp_content_type("application/json")
         |> Conn.send_resp(response_code, response)
     after

lib/rig_inbound_gateway/api_proxy/handler/kafka.ex

@@ -44,12 +44,12 @@ defmodule RigInboundGateway.ApiProxy.Handler.Kafka do
              any()}
   def kafka_handler(message, headers) do
     case ResponseFromParser.parse(headers, message) do
-      {deserialized_pid, response_code, response} ->
+      {deserialized_pid, response_code, response, extra_headers} ->
         Logger.debug(fn ->
           "HTTP response via Kafka to #{inspect(deserialized_pid)}: #{inspect(message)}"
         end)
 
-        send(deserialized_pid, {:response_received, response, response_code})
+        send(deserialized_pid, {:response_received, response, response_code, extra_headers})
 
       err ->
         Logger.warn(fn -> "Parse error #{inspect(err)} for #{inspect(message)}" end)
@@ -257,7 +257,7 @@ defmodule RigInboundGateway.ApiProxy.Handler.Kafka do
     conf = config()
 
     receive do
-      {:response_received, response, response_code} ->
+      {:response_received, response, response_code, extra_headers} ->
         ProxyMetrics.count_proxy_request(
           conn.method,
           conn.request_path,
@@ -268,6 +268,9 @@ defmodule RigInboundGateway.ApiProxy.Handler.Kafka do
 
         conn
         |> Tracing.Plug.put_resp_header(Tracing.context())
+        |> Map.update!(:resp_headers, fn existing_headers ->
+          existing_headers ++ Map.to_list(extra_headers)
+        end)
         |> Conn.put_resp_content_type("application/json")
         |> Conn.send_resp(response_code, response)
     after

lib/rig_inbound_gateway/api_proxy/response_from_parser.ex

@@ -21,14 +21,18 @@ defmodule RigInboundGateway.ApiProxy.ResponseFromParser do
          {:ok, correlation_id} <- Map.fetch(headers_map, "rig-correlation"),
          {:ok, deserialized_pid} <- Codec.deserialize(correlation_id),
          {:ok, raw_response_code} <- Map.fetch(headers_map, "rig-response-code"),
+         # forward extra haeders such as content-type
+         extra_headers <- Map.drop(headers_map, ["rig-correlation", "rig-response-code"]),
          # convert status code to int if needed, HTTP headers can't have number as a value
          response_code <- to_int(raw_response_code),
          response_body <- try_encode(message) do
       Logger.debug(fn ->
-        "Parsed binary HTTP response: body=#{inspect(response_body)}, code=#{inspect(response_code)}"
+        "Parsed binary HTTP response: body=#{inspect(response_body)}, code=#{
+          inspect(response_code)
+        }, headers=#{inspect(extra_headers)}"
       end)
 
-      {deserialized_pid, response_code, response_body}
+      {deserialized_pid, response_code, response_body, extra_headers}
     else
       err -> err
     end
@@ -51,6 +55,7 @@ defmodule RigInboundGateway.ApiProxy.ResponseFromParser do
   # ---
 
   defp to_int(value) when is_integer(value), do: value
+
   defp to_int(string) when is_binary(string) do
     String.to_integer(string)
   rescue

test/rig_tests/proxy/response_from/async_http_test.exs

@@ -77,13 +77,17 @@ defmodule RigTests.Proxy.ResponseFrom.AsyncHttpTest do
 
     # The client calls the proxy endpoint:
     request_url = rig_proxy_url <> endpoint_path
-    %HTTPoison.Response{status_code: res_status, body: res_body} = HTTPoison.get!(request_url)
+
+    %HTTPoison.Response{status_code: res_status, body: res_body, headers: headers} =
+      HTTPoison.get!(request_url)
 
     # Now we can assert that...
     # ...the fake backend service has been called:
     assert FakeServer.hits() == 1
     # ...the connection is closed and the status is OK:
     assert res_status == 201
+    # ...extra headers are present
+    assert Enum.member?(headers, {"content-type", "application/json; charset=utf-8"})
     # ...but the client got the response sent to the HTTP internal endpoint:
     assert Jason.decode!(res_body) == async_response
   end

test/rig_tests/proxy/response_from/kafka_test.exs

@@ -108,13 +108,16 @@ defmodule RigTests.Proxy.ResponseFrom.KafkaTest do
     # The client calls the proxy endpoint:
     request_url = rig_proxy_url <> endpoint_path
 
-    %HTTPoison.Response{status_code: res_status, body: res_body} = HTTPoison.get!(request_url)
+    %HTTPoison.Response{status_code: res_status, body: res_body, headers: headers} =
+      HTTPoison.get!(request_url)
 
     # Now we can assert that...
     # ...the fake backend service has been called:
     assert FakeServer.hits() == 1
     # ...the connection is closed and the status is OK:
     assert res_status == 201
+    # ...extra headers are present
+    assert Enum.member?(headers, {"content-type", "application/json; charset=utf-8"})
     # ...but the client got the response sent to the Kafka topic:
     assert Jason.decode!(res_body) == async_response