221 - added new /v3 internal api to support response code

Author: mmacai

CHANGELOG.md

@@ -13,6 +13,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Added validation for reverse proxy configuration. Now it crashes RIG on start when configuration is not valid or returns `400` when using REST API to update configuration. [#277](https://github.com/Accenture/reactive-interaction-gateway/issues/277)
 - Added basic distributed tracing support in [W3C Trace Context specification](https://www.w3.org/TR/trace-context/) with Jaeger and Openzipkin exporters. RIG opens a span at the API Gateway and emits trace context in Cloud Events following the [distributed tracing spec](https://github.com/cloudevents/spec/blob/v1.0/extensions/distributed-tracing.md). [#281](https://github.com/Accenture/reactive-interaction-gateway/issues/281)
 - Added possibility to set response code for `response_from` messages in reverse proxy (`kafka` and `http_async`). [#321](https://github.com/Accenture/reactive-interaction-gateway/pull/321)
+- Added new version - `v3` - for internal endpoints to support response code in the `/responses` endpoint
 
 ### Changed
 

config/rig_api/config.exs

@@ -45,6 +45,7 @@ config :phoenix, :serve_endpoints, true
 
 config :rig, RigApi.V1.APIs, rig_proxy: RigInboundGateway.Proxy
 config :rig, RigApi.V2.APIs, rig_proxy: RigInboundGateway.Proxy
+config :rig, RigApi.V3.APIs, rig_proxy: RigInboundGateway.Proxy
 
 config :rig, :event_filter, Rig.EventFilter
 

config/rig_api/test.exs

@@ -10,5 +10,6 @@ config :rig, RigApi.Endpoint,
 
 config :rig, RigApi.V1.APIs, rig_proxy: RigInboundGateway.ProxyMock
 config :rig, RigApi.V2.APIs, rig_proxy: RigInboundGateway.ProxyMock
+config :rig, RigApi.V3.APIs, rig_proxy: RigInboundGateway.ProxyMock
 
 config :rig, :event_filter, Rig.EventFilterMock

lib/rig_api/router.ex

@@ -75,6 +75,32 @@ defmodule RigApi.Router do
     end
   end
 
+  scope "/v3", RigApi.V3 do
+    scope "/apis" do
+      pipe_through(:body_parser)
+      get("/", APIs, :list_apis)
+      post("/", APIs, :add_api)
+      get("/:id", APIs, :get_api_detail)
+      put("/:id", APIs, :update_api)
+      delete("/:id", APIs, :deactivate_api)
+    end
+
+    scope "/messages" do
+      post("/", Messages, :publish)
+    end
+
+    scope "/responses" do
+      pipe_through(:body_parser)
+      resources("/", Responses, only: [:create])
+    end
+
+    scope "/session-blacklist" do
+      pipe_through(:body_parser)
+      post("/", SessionBlacklist, :blacklist_session)
+      get("/:session_id", SessionBlacklist, :check_status)
+    end
+  end
+
   def swagger_info do
     %{
       info: %{

lib/rig_api/v1/responses.ex

@@ -5,7 +5,8 @@ defmodule RigApi.V1.Responses do
   use RigApi, :controller
   use PhoenixSwagger
 
-  alias RigInboundGateway.ApiProxy.ResponseFromParser
+  alias Rig.Connection.Codec
+  alias RigCloudEvents.CloudEvent
 
   @prefix "/v1"
 
@@ -32,19 +33,20 @@ defmodule RigApi.V1.Responses do
 
   @doc """
   Accepts message to be sent to correlated HTTP process.
-
   Note that body has to contain following field `"rig": { "correlation": "_id_" }`.
   """
-  def create(%{req_headers: req_headers} = conn, message) do
-    case ResponseFromParser.parse(req_headers, message) do
-      {deserialized_pid, response_code, response} ->
-        Logger.debug(fn ->
-          "HTTP response via internal HTTP to #{inspect(deserialized_pid)}: #{inspect(message)}"
-        end)
-
-        send(deserialized_pid, {:response_received, response, response_code})
-        send_resp(conn, :accepted, "message sent to correlated reverse proxy request")
-
+  def create(conn, message) do
+    with {:ok, cloud_event} <- CloudEvent.parse(message),
+         {:ok, rig_metadata} <- Map.fetch(message, "rig"),
+         {:ok, correlation_id} <- Map.fetch(rig_metadata, "correlation"),
+         {:ok, deserialized_pid} <- Codec.deserialize(correlation_id) do
+      Logger.debug(fn ->
+        "HTTP response via internal HTTP to #{inspect(deserialized_pid)}: #{inspect(message)}"
+      end)
+
+      send(deserialized_pid, {:response_received, cloud_event.json})
+      send_resp(conn, :accepted, "message sent to correlated reverse proxy request")
+    else
       err ->
         Logger.warn(fn -> "Parse error #{inspect(err)} for #{inspect(message)}" end)
 

lib/rig_api/v2/responses.ex

@@ -5,7 +5,8 @@ defmodule RigApi.V2.Responses do
   use RigApi, :controller
   use PhoenixSwagger
 
-  alias RigInboundGateway.ApiProxy.ResponseFromParser
+  alias Rig.Connection.Codec
+  alias RigCloudEvents.CloudEvent
 
   @prefix "/v2"
 
@@ -32,19 +33,20 @@ defmodule RigApi.V2.Responses do
 
   @doc """
   Accepts message to be sent to correlated HTTP process.
-
   Note that body has to contain following field `"rig": { "correlation": "_id_" }`.
   """
-  def create(%{req_headers: req_headers} = conn, message) do
-    case ResponseFromParser.parse(req_headers, message) do
-      {deserialized_pid, response_code, response} ->
-        Logger.debug(fn ->
-          "HTTP response via internal HTTP to #{inspect(deserialized_pid)}: #{inspect(message)}"
-        end)
-
-        send(deserialized_pid, {:response_received, response, response_code})
-        send_resp(conn, :accepted, "message sent to correlated reverse proxy request")
-
+  def create(conn, message) do
+    with {:ok, cloud_event} <- CloudEvent.parse(message),
+         {:ok, rig_metadata} <- Map.fetch(message, "rig"),
+         {:ok, correlation_id} <- Map.fetch(rig_metadata, "correlation"),
+         {:ok, deserialized_pid} <- Codec.deserialize(correlation_id) do
+      Logger.debug(fn ->
+        "HTTP response via internal HTTP to #{inspect(deserialized_pid)}: #{inspect(message)}"
+      end)
+
+      send(deserialized_pid, {:response_received, cloud_event.json})
+      send_resp(conn, :accepted, "message sent to correlated reverse proxy request")
+    else
       err ->
         Logger.warn(fn -> "Parse error #{inspect(err)} for #{inspect(message)}" end)