Merge pull request #380 from rbino/trigger-engine-prefetch-count

trigger_engine: add AMQP prefetch count

CHANGELOG.md

@@ -8,6 +8,8 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 ### Added
 - [data_updater_plant] Add `DATA_UPDATER_PLANT_AMQP_DATA_QUEUE_TOTAL_COUNT` environment variable,
   this must be equal to the total number of queues in the Astarte instance.
+- [trigger_engine] Add `TRIGGER_ENGINE_AMQP_PREFETCH_COUNT` environment variable to set the
+  prefetech count of AMQPEventsConsumer, avoiding excessive memory usage.
 
 ### Fixed
 - Wait for schema_version agreement before applying any schema change (such as creating tables or a

apps/astarte_trigger_engine/config/astarte_trigger_engine.schema.exs

@@ -97,6 +97,15 @@ See the moduledoc for `Conform.Schema.Validator` for more details and examples.
       hidden: false,
       to: "astarte_trigger_engine.amqp_consumer_options.virtual_host"
     ],
+    "amqp_consumer_options.virtual_host": [
+      commented: true,
+      datatype: :integer,
+      default: 300,
+      env_var: "TRIGGER_ENGINE_AMQP_PREFETCH_COUNT",
+      doc: "Prefetch count for the AMQP consumer",
+      hidden: false,
+      to: "astarte_trigger_engine.amqp_prefetch_count"
+    ],
     amqp_events_queue_name: [
       commented: true,
       datatype: :binary,

apps/astarte_trigger_engine/lib/astarte_trigger_engine/amqp_events_consumer.ex

@@ -113,6 +113,7 @@ defmodule Astarte.TriggerEngine.AMQPEventsConsumer do
   defp connect() do
     with {:ok, conn} <- Connection.open(Config.amqp_consumer_options()),
          {:ok, chan} <- Channel.open(conn),
+         :ok <- Basic.qos(chan, prefetch_count: Config.amqp_prefetch_count()),
          :ok <- Exchange.declare(chan, Config.events_exchange_name(), :direct, durable: true),
          {:ok, _queue} <- Queue.declare(chan, Config.events_queue_name(), durable: true),
          :ok <-

apps/astarte_trigger_engine/lib/astarte_trigger_engine/config.ex

@@ -28,6 +28,13 @@ defmodule Astarte.TriggerEngine.Config do
     Application.get_env(:astarte_trigger_engine, :amqp_consumer_options, [])
   end
 
+  @doc """
+  Returns the AMQP prefetch count
+  """
+  def amqp_prefetch_count do
+    Application.get_env(:astarte_trigger_engine, :amqp_prefetch_count, 300)
+  end
+
   @doc """
   Returns the events name of the exchange on which events are published
   """