Merge pull request #318 from Accenture/300-async-service-send-response-directly

make it possible to return cached response

Author: kevinbader

docs/api-gateway.md

@@ -180,7 +180,7 @@ The endpoint expects the following request format:
 ### Wait for response
 
 Sometimes it makes sense to provide a simple request-response API to something that runs asynchronously on the backend. For example, let's say there's a ticket reservation process that takes 10 seconds in total and involves three different services that communicate via message passing. For an external client, it may be simpler to wait 10 seconds for the response instead of polling for a response every other second.
-A behavior like this can be configured using an endpoints' `response_from` property. When set to `kafka`, the response to the request is not taken from the `target` (e.g., for `target` = `http` this means the backend's HTTP response is ignored), but instead it's read from a Kafka topic. In order to enable RIG to correlate the response from the topic with the original request, RIG adds a correlation ID to the request (using a query parameter in case of `target` = `http`, or backed into the produced CloudEvent otherwise). **Backend services that work with the request need to include that correlation ID in their response; otherwise, RIG won't be able to forward it to the client (and times out).**
+A behavior like this can be configured using an endpoints' `response_from` property. When set to `kafka`, the response to the request is not necessarily taken from the `target`, e.g., for `target` = `http` this means the backend's HTTP response might be ignored - it's the responsibility of the backend service where to read the response from: If the backend returns with the HTTP code `202 Accepted`, the response will be read from a Kafka topic. If the backend returns a different HTTP code (can be `200` or `400` or whatever makes sense), only then the response will be read synchronously from the http target directly (this allows the backend to return cached responses). In order to enable RIG to correlate the response from the kafka topic with the original request, RIG adds a correlation ID to the request (using a query parameter in case of `target` = `http`, or baked into the produced CloudEvent otherwise). Backend services that work with the request need to include that correlation ID in their response; otherwise, RIG won't be able to forward it to the client (and times out).
 
 Configuration of such API endpoint might look like this:
 

lib/rig_inbound_gateway/api_proxy/handler/http.ex

@@ -85,10 +85,19 @@ defmodule RigInboundGateway.ApiProxy.Handler.Http do
   defp handle_response(conn, res, response_from)
 
   defp handle_response(conn, res, "http"),
-    do: send_or_chunk_response(conn, res)
+    do: send_or_chunk_response(conn, res, "http")
 
-  defp handle_response(conn, _, response_from),
-    do: wait_for_response(conn, response_from)
+  defp handle_response(conn, res, response_from),
+    do: respond_directly_or_wait_for_response(conn, res, response_from)
+
+  # ---
+
+  defp respond_directly_or_wait_for_response(conn, res, response_from) do
+    case res.status_code do
+      202 -> wait_for_response(conn, response_from)
+      _ -> send_or_chunk_response(conn, res, response_from)
+    end
+  end
 
   # ---
 
@@ -192,14 +201,15 @@ defmodule RigInboundGateway.ApiProxy.Handler.Http do
 
   # ---
 
-  @spec send_or_chunk_response(Plug.Conn.t(), HTTPoison.Response.t()) :: Plug.Conn.t()
+  @spec send_or_chunk_response(Plug.Conn.t(), HTTPoison.Response.t(), String.t()) :: Plug.Conn.t()
   defp send_or_chunk_response(
          conn,
          %HTTPoison.Response{
            headers: headers,
            status_code: status_code,
            body: body
-         }
+         },
+         response_from
        ) do
     headers =
       headers
@@ -208,8 +218,7 @@ defmodule RigInboundGateway.ApiProxy.Handler.Http do
 
     conn = %{conn | resp_headers: headers}
 
-    # only possibility for "response_from" = "http", therefore hardcoded here
-    ProxyMetrics.count_proxy_request(conn.method, conn.request_path, "http", "http", "ok")
+    ProxyMetrics.count_proxy_request(conn.method, conn.request_path, "http", response_from, "ok")
 
     headers
     |> Map.new()

test/rig_tests/proxy/response_from/async_http_test.exs

@@ -27,8 +27,10 @@ defmodule RigTests.Proxy.ResponseFrom.AsyncHttpTest do
     end)
   end
 
-  test_with_server "Given response_from is set to http_async and response is in binary mode, the http response is taken from the internal HTTP endpoint." do
-    test_name = "proxy-http-response-from-http-internal-binary"
+  # Given response_from=http_async, when the backend responds with 202, RIG doesn't forward that 202-response.
+  # Instead, RIG expects the actual response to be submitted to its API, identified by the correlation ID RIG has added to the forwarded request.
+  test_with_server "Given response_from=http_async, when the backend responds with 202, the http response is taken from the internal HTTP endpoint" do
+    test_name = "proxy-http-response-from-http-internal"
 
     api_id = "mock-#{test_name}-api"
     endpoint_id = "mock-#{test_name}-endpoint"
@@ -43,7 +45,7 @@ defmodule RigTests.Proxy.ResponseFrom.AsyncHttpTest do
       |> put_req_header("content-type", "application/json;charset=utf-8")
       |> post("/v3/responses", Jason.encode!(async_response))
 
-      Response.ok!(sync_response, %{"content-type" => "application/json"})
+      Response.accepted!(sync_response, %{"content-type" => "application/json"})
     end)
 
     # We register the endpoint with the proxy:
@@ -112,7 +114,7 @@ defmodule RigTests.Proxy.ResponseFrom.AsyncHttpTest do
       assert res_status == 400
       assert res_body == "Failed to parse request body: {:error, {:not_an_integer, \"abc201\"}}"
 
-      Response.ok!(sync_response, %{"content-type" => "application/json"})
+      Response.accepted!(sync_response, %{"content-type" => "application/json"})
     end)
 
     # We register the endpoint with the proxy:
@@ -153,4 +155,144 @@ defmodule RigTests.Proxy.ResponseFrom.AsyncHttpTest do
     # ...the fake backend service has been called:
     assert FakeServer.hits() == 1
   end
+
+  test_with_server "Given response_from=http_async, when the backend responds with 200, RIG forwards this 200-response (and does not expect an asynchronous response for this request)." do
+    test_name = "proxy-http-response-synchronous"
+
+    api_id = "mock-#{test_name}-api"
+    endpoint_id = "mock-#{test_name}-endpoint"
+    endpoint_path = "/#{endpoint_id}"
+    sync_response = %{"message" => "this is the sync response that reaches the client"}
+    async_response = %{"this response" => "the client never sees this response"}
+
+    route(endpoint_path, fn %{query: %{"correlation" => correlation_id}} ->
+      event =
+        Jason.encode!(%{
+          specversion: "0.2",
+          type: "rig.async-response",
+          source: "fake-service",
+          id: "1",
+          rig: %{correlation: correlation_id},
+          data: async_response
+        })
+
+      build_conn()
+      |> put_req_header("content-type", "application/json;charset=utf-8")
+      |> post("/v2/responses", event)
+
+      Response.ok!(sync_response, %{"content-type" => "application/json"})
+    end)
+
+    # We register the endpoint with the proxy:
+    rig_api_url = "http://localhost:#{@api_port}/v2/apis"
+    rig_proxy_url = "http://localhost:#{@proxy_port}"
+
+    body =
+      Jason.encode!(%{
+        id: api_id,
+        name: "Mock API",
+        version_data: %{
+          default: %{
+            endpoints: [
+              %{
+                id: endpoint_id,
+                type: "http",
+                method: "GET",
+                path: endpoint_path,
+                response_from: "http_async"
+              }
+            ]
+          }
+        },
+        proxy: %{
+          target_url: "localhost",
+          port: FakeServer.port()
+        }
+      })
+
+    headers = [{"content-type", "application/json"}]
+    HTTPoison.post!(rig_api_url, body, headers)
+
+    # 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)
+
+    # 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 == 200
+    # ...the client receives the synchronous http response:
+    assert Jason.decode!(res_body) == sync_response
+  end
+
+  test_with_server "Given response_from=http_async, when the backend responds with 400, RIG forwards this 400-response (and does not expect an asynchronous response for this request)." do
+    test_name = "proxy-http-no-response"
+
+    api_id = "mock-#{test_name}-api"
+    endpoint_id = "mock-#{test_name}-endpoint"
+    endpoint_path = "/#{endpoint_id}"
+    sync_response = "Bad request from the test endpoint"
+    async_response = %{"this response" => "the client never sees this response"}
+
+    route(endpoint_path, fn %{query: %{"correlation" => correlation_id}} ->
+      event =
+        Jason.encode!(%{
+          specversion: "0.2",
+          type: "rig.async-response",
+          source: "fake-service",
+          id: "1",
+          rig: %{correlation: correlation_id},
+          data: async_response
+        })
+
+      build_conn()
+      |> put_req_header("content-type", "application/json;charset=utf-8")
+      |> post("/v2/responses", event)
+
+      Response.bad_request!(sync_response)
+    end)
+
+    # We register the endpoint with the proxy:
+    rig_api_url = "http://localhost:#{@api_port}/v2/apis"
+    rig_proxy_url = "http://localhost:#{@proxy_port}"
+
+    body =
+      Jason.encode!(%{
+        id: api_id,
+        name: "Mock API",
+        version_data: %{
+          default: %{
+            endpoints: [
+              %{
+                id: endpoint_id,
+                type: "http",
+                method: "GET",
+                path: endpoint_path,
+                response_from: "http_async"
+              }
+            ]
+          }
+        },
+        proxy: %{
+          target_url: "localhost",
+          port: FakeServer.port()
+        }
+      })
+
+    headers = [{"content-type", "application/json"}]
+    HTTPoison.post!(rig_api_url, body, headers)
+
+    # 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)
+
+    # 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 == 400
+    # ...the client does not get any response:
+    assert res_body == sync_response
+  end
 end