Process one message at a time

Make agents process only one message at a time and remove
`bulk_receive` functionality.

Processing multiple messages in a go was mostly a performance
improvement. Semantically it is much cleaner when `receive` processes a
single message so that it matches transaction boundaries.

This also opens the doors to solving existing concurrency issues.

Breaking change:

CSV Agent did use the functionality, where it would process several
messages (rows) in one go and emit a single CSV message. This behavior
wasn't deterministic and from now on it is limited to producing one CSV
message from one `data` message containing multiple rows.

(Custom and remote agents already conform to a single message per
receive API).

Author: Vidas P

CHANGELOG.md

@@ -6,9 +6,19 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## [current]
+### Changed
+- Semantics of message delivery got different: agents now process a single
+  message at a time, thus matching transactional boundaries.
+- BREAKING: CSV Agent in 'serialize' mode handles only a single message at a
+  time and can't aggregate multiple 'row' messages into a single CSV. Old
+  behaviour wasn't exactly deterministic, suggested way would be to use
+  `Digest Agent` to aggregate messages and then serialize them with
+  `CSV Agent`.
+
 ### Fixed
 - Update nokogiri (CVE-2020-7595)
 
+
 ## [0.9.6] - 2019-12-09
 ### Added
 - Support for remote agents. Custom agents can now be written in any

app/concerns/dry_runnable.rb

@@ -26,7 +26,7 @@ def dry_run!(message = nil)
       @dry_run_logger.info('Dry Run started')
       if message
         raise 'This agent cannot receive an message!' unless can_receive_messages?
-        receive([message])
+        receive(message)
       else
         check
       end

app/concerns/liquid_interpolatable.rb

@@ -43,14 +43,12 @@ def interpolation_context
   # Take the given object as "self" in the current interpolation
   # context while running a given block.
   #
-  # The most typical use case for this is to evaluate options for each
-  # received message like this:
+  # The most typical use case for this is to evaluate
+  # options for a received message like this:
   #
-  #     def receive(incoming_messages)
-  #       incoming_messages.each do |message|
-  #         interpolate_with(message) do
+  #     def receive(message)
+  #       interpolate_with(message) do
   #           # Handle each message based on "interpolated" options.
-  #         end
   #       end
   #     end
   def interpolate_with(self_object)

app/concerns/sortable_messages.rb

@@ -95,13 +95,11 @@ def check
       end
     end
 
-    def receive(incoming_messages)
+    def receive(message)
       return super unless messages_order || include_sort_info?
-      # incoming messages should be processed sequentially
-      incoming_messages.each do |message|
-        sorting_messages do
-          super([message])
-        end
+
+      sorting_messages do
+        super(message)
       end
     end
 

app/jobs/agent_receive_job.rb

@@ -2,11 +2,11 @@ class AgentReceiveJob < ActiveJob::Base
   # Given an Agent id and an array of Message ids, load the Agent, call #receive on it with the Message objects, and then
   # save it with an updated `last_receive_at` timestamp.
   # rubocop:disable Style/RescueStandardError
-  def perform(agent_id, message_ids)
+  def perform(agent_id, message_id)
     agent = Agent.find(agent_id)
     begin
       return if agent.unavailable?
-      agent.receive(Message.where(id: message_ids).order(:id))
+      agent.receive(Message.find(message_id))
       agent.last_receive_at = Time.now
       agent.save!
     rescue => e

app/models/agent.rb

@@ -95,7 +95,7 @@ def default_options
     {}
   end
 
-  def receive(messages)
+  def receive(message)
     # Implement me in your subclass of Agent.
   end
 
@@ -238,10 +238,6 @@ def can_dry_run?
     self.class.can_dry_run?
   end
 
-  def no_bulk_receive?
-    self.class.no_bulk_receive?
-  end
-
   def log(message, options = {})
     AgentLog.log_for_agent(self, message, options)
   end
@@ -371,14 +367,6 @@ def can_dry_run?
       !!@can_dry_run
     end
 
-    def no_bulk_receive!
-      @no_bulk_receive = true
-    end
-
-    def no_bulk_receive?
-      !!@no_bulk_receive
-    end
-
     def gem_dependency_check
       @gem_dependencies_checked = true
       @gem_dependencies_met = yield
@@ -424,11 +412,7 @@ def receive!(options = {})
           agent.update_attribute :last_checked_message_id, message_ids.max
           # rubocop:enable Rails/SkipsModelValidations
 
-          if agent.no_bulk_receive?
-            message_ids.each { |message_id| Agent.async_receive(agent.id, [message_id]) }
-          else
-            Agent.async_receive(agent.id, message_ids)
-          end
+          message_ids.each { |message_id| Agent.async_receive(agent.id, message_id) }
         end
 
         {
@@ -440,8 +424,8 @@ def receive!(options = {})
 
     # This method will enqueue an AgentReceiveJob job. It accepts Agent and Message ids instead of a literal ActiveRecord
     # models because it is preferable to serialize jobs with ids.
-    def async_receive(agent_id, message_ids)
-      AgentReceiveJob.perform_later(agent_id, message_ids)
+    def async_receive(agent_id, message_id)
+      AgentReceiveJob.perform_later(agent_id, message_id)
     end
 
     # Given a schedule name, run `check` via `bulk_check` on all Agents with that schedule.

app/models/agents/attribute_difference_agent.rb

@@ -49,10 +49,8 @@ def validate_options
       end
     end
 
-    def receive(incoming_messages)
-      incoming_messages.each do |message|
-        handle(interpolated(message), message)
-      end
+    def receive(message)
+      handle(interpolated(message), message)
     end
 
     private

app/models/agents/buffer_agent.rb

@@ -55,16 +55,14 @@ def validate_options
     end
     # rubocop:enable Metrics/CyclomaticComplexity
 
-    def receive(incoming_messages)
-      incoming_messages.each do |message|
-        memory['message_ids'] ||= []
-        memory['message_ids'] << message.id
-        if memory['message_ids'].length > interpolated['max_messages'].to_i
-          if options['keep'] == 'newest'
-            memory['message_ids'].shift
-          else
-            memory['message_ids'].pop
-          end
+    def receive(message)
+      memory['message_ids'] ||= []
+      memory['message_ids'] << message.id
+      if memory['message_ids'].length > interpolated['max_messages'].to_i
+        if options['keep'] == 'newest'
+          memory['message_ids'].shift
+        else
+          memory['message_ids'].pop
         end
       end
     end

app/models/agents/change_detector_agent.rb

@@ -37,12 +37,10 @@ def validate_options
       end
     end
 
-    def receive(incoming_messages)
-      incoming_messages.each do |message|
-        interpolation_context.stack do
-          interpolation_context['last_property'] = last_property
-          handle(interpolated(message), message)
-        end
+    def receive(message)
+      interpolation_context.stack do
+        interpolation_context['last_property'] = last_property
+        handle(interpolated(message), message)
       end
     end
 

app/models/agents/commander_agent.rb

@@ -32,11 +32,9 @@ def check
       control!
     end
 
-    def receive(incoming_messages)
-      incoming_messages.each do |message|
-        interpolate_with(message) do
-          control!
-        end
+    def receive(message)
+      interpolate_with(message) do
+        control!
       end
     end
   end