221 - cleanup

Author: mmacai

CHANGELOG.md

@@ -12,14 +12,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Support publishing events consumed from [NATS](https://nats.io) topics. See the [documentation](https://accenture.github.io/reactive-interaction-gateway/docs/event-streams.html#nats) for how to get started. [#297](https://github.com/Accenture/reactive-interaction-gateway/issues/297)
 - 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. [#321](https://github.com/Accenture/reactive-interaction-gateway/pull/321)
+- 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)
 
 ### Changed
 
 - Incorporated [cloudevents-ex](https://github.com/kevinbader/cloudevents-ex) to handle binary and structured modes for [Kafka protocol binding](https://github.com/cloudevents/spec/blob/v1.0/kafka-protocol-binding.md) in a proper way. This introduces some **breaking changes**:
   - Binary mode is now using `ce_` prefix for CloudEvents context attribute headers, before it was `ce-` - done according to the [Kafka protocol binding](https://github.com/cloudevents/spec/blob/v1.0/kafka-protocol-binding.md)
 - Change above affects also `"response_from": "kafka"` proxy functionality. RIG will forward to clients only Kafka body, no headers. This means, when using binary mode, clients receive only the data part, no CloudEvents context attributes.
-- Changed `response_from` handler to expect a message in binary format, **NOT** a cloud event. [#321](https://github.com/Accenture/reactive-interaction-gateway/pull/321)
+- Changed `response_from` handler to expect a message in binary format, **NOT** a cloud event (`kafka` and `http_async`). [#321](https://github.com/Accenture/reactive-interaction-gateway/pull/321)
 
 ### Fixed
 

config/rig_inbound_gateway/config.exs

@@ -172,9 +172,6 @@ config :rig, RigInboundGatewayWeb.V1.SubscriptionController, cors: cors
 config :rig, RigInboundGatewayWeb.V1.SSE, cors: cors
 config :rig, RigInboundGatewayWeb.V1.LongpollingController, cors: cors
 
-config :rig, RigInboundGateway.ApiProxy.ResponseFromParser,
-  schema_registry_host: {:system, "KAFKA_SCHEMA_REGISTRY_HOST", nil}
-
 # Import environment specific config. This must remain at the bottom
 # of this file so it overrides the configuration defined above.
 import_config "#{Mix.env()}.exs"

docs/api-gateway.md

@@ -214,8 +214,6 @@ Configuration of such API endpoint might look like this:
 
 `http_async` means that correlated response has to be sent to internal `:4010/v2/responses` `POST` endpoint.
 
-`response_from="kafka"` will try to decode Avro encoded message.
-
 #### Supported formats
 
 All `response_from` options support only binary mode.

lib/rig_inbound_gateway/api_proxy/handler/kafka.ex

@@ -43,14 +43,14 @@ defmodule RigInboundGateway.ApiProxy.Handler.Kafka do
           | {:error, %{:__exception__ => true, :__struct__ => atom(), optional(atom()) => any()},
              any()}
   def kafka_handler(message, headers) do
-    with {deserialized_pid, response_code, response} <-
-           ResponseFromParser.parse(headers, message) do
-      Logger.debug(fn ->
-        "HTTP response via Kafka to #{inspect(deserialized_pid)}: #{inspect(message)}"
-      end)
+    case ResponseFromParser.parse(headers, message) do
+      {deserialized_pid, response_code, response} ->
+        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})
-    else
       err ->
         Logger.warn(fn -> "Parse error #{inspect(err)} for #{inspect(message)}" end)
         :ignore

lib/rig_inbound_gateway/api_proxy/response_from_parser.ex

@@ -6,8 +6,6 @@ defmodule RigInboundGateway.ApiProxy.ResponseFromParser do
 
   require Logger
 
-  use Rig.Config, [:schema_registry_host]
-
   alias Rig.Connection.Codec
 
   # ---