Eh main/feature/consumer

lib/pipefy_message/consumer.rb

@@ -14,6 +14,7 @@ module Consumer
     include PipefyMessage::Logging
     include PipefyMessage::Providers::Errors
     include PipefyMessage::Providers::BrokerResolver
+
     ##
     # Default options for consumer setup.
     def self.default_consumer_options
@@ -47,33 +48,15 @@ module ClassMethods
       # Merges default worker options with the hash passed as
       # an argument. The latter takes precedence.
       def options(opts = {})
-        @options = Consumer.default_consumer_options.merge(opts)
-        @options.each do |k, v|
-          singleton_class.class_eval { attr_accessor k }
-          send("#{k}=", v)
-        end
-
-        logger.debug({
-                       options_set: @options,
-                       message_text: "Set #{name} options to options_set"
-                     })
-      end
-
-      ##
-      # Sets broker-specific options to be passed to the broker's
-      # constructor.
-      def broker_options(opts = {})
-        @broker_opts = opts
+        @consumer_options = Consumer.default_consumer_options.merge(opts)
       end
 
       ##
       # Initializes and returns an instance of a broker for
       # the provider specified in the class options.
       def build_consumer_instance
-        options if @broker.nil?
-        broker_options if @broker_opts.nil?
-        consumer_map = resolve_broker(@broker, "consumer")
-        consumer_map[:class_name].constantize.new(@broker_opts)
+        consumer_map = resolve_broker(@consumer_options[:broker], "consumer")
+        consumer_map[:class_name].constantize.new(@consumer_options)
       end
 
       ##
@@ -85,27 +68,27 @@ def build_consumer_instance
       def process_message
         start = Time.now
         obj = new
-        logger.info({
-                      message_text: "Calling poller for #{@broker} object"
-                    })
 
-        build_consumer_instance.poller do |message|
+        logger.info({ message_text: "Calling consumer poller" })
+
+        build_consumer_instance.poller do |payload|
+          logger.info({
+                        message_text: "Message received by poller to be processed by consumer",
+                        received_message: payload
+                      })
+
+          obj.perform(payload["Message"])
+
+          elapsed_time = (Time.now - start) * 1000.0
           logger.info({
-                        message_text: "Message received by #{@broker} poller to be processed by consumer",
-                        received_message: message
+                        duration_ms: elapsed_time,
+                        message_text: "Message received by consumer poller, processed " \
+                                      "in #{elapsed_time} milliseconds"
                       })
-          obj.perform(message)
         end
-      rescue PipefyMessage::Providers::Errors::ResourceError => e # (any others?)
+      rescue PipefyMessage::Providers::Errors::ResourceError => e
+        logger.error("Failed to process message, details #{e.inspect}")
         raise e
-      ensure
-        elapsed_time = (Time.now - start) * 1000.0
-        logger.info({
-                      duration_ms: elapsed_time,
-                      message_text: "Message received by #{@broker}" \
-                                    "poller processed by #{name} worker" \
-                                    "in #{elapsed_time} milliseconds"
-                    })
       end
     end
   end

lib/pipefy_message/providers/aws_client/sqs_broker.rb

@@ -32,13 +32,16 @@ def initialize(opts = {})
         # Initiates SQS queue polling, with wait_time_seconds as given
         # in the initial configuration.
         def poller
-          logger.debug(build_log_hash("Initiating SQS polling on queue #{@config[:queue_name]}"))
+          logger.info(build_log_hash("Initiating SQS polling on queue #{@config[:queue_name]}"))
 
           @poller.poll(wait_time_seconds: @config[:wait_time_seconds]) do |received_message|
             logger.debug(build_log_hash("Message received by SQS poller on queue #{@config[:queue_name]}"))
 
             payload = JSON.parse(received_message.body)
             yield(payload)
+
+          rescue StandardError => e
+            raise PipefyMessage::Providers::Errors::ResourceError, e.message
           end
         end
 

spec/consumer_spec.rb

@@ -10,7 +10,7 @@ def poller
   end
 end
 
-class TestWorker
+class TestConsumer
   include PipefyMessage::Consumer
   options broker: "aws", queue_name: "pipefy-local-queue"
 
@@ -20,49 +20,53 @@ def perform(message)
 end
 
 RSpec.describe PipefyMessage::Consumer do
+  before do
+    ENV["ENABLE_AWS_CLIENT_CONFIG"] = "true"
+    ENV["AWS_CLI_STUB_RESPONSE"] = "true"
+  end
   describe "#perform" do
     context "successful polling" do
       it "should call #perform from child instance when #process_message is called" do
         mock_broker = instance_double("MockBroker")
         allow(mock_broker).to receive(:poller).with(no_args)
 
-        allow(TestWorker).to receive(:build_consumer_instance).and_return(mock_broker)
+        allow(TestConsumer).to receive(:build_consumer_instance).and_return(mock_broker)
 
-        TestWorker.process_message
+        TestConsumer.process_message
         expect(mock_broker).to have_received(:poller)
       end
     end
 
     context "polling failure" do
       it "should call #perform from child instance when #process_message is called" do
-        allow(TestWorker).to receive(:build_consumer_instance).and_return(MockBrokerFail.new)
-        expect { TestWorker.process_message }.to raise_error(PipefyMessage::Providers::Errors::ResourceError)
+        allow(TestConsumer).to receive(:build_consumer_instance).and_return(MockBrokerFail.new)
+        expect { TestConsumer.process_message }.to raise_error(PipefyMessage::Providers::Errors::ResourceError)
       end
     end
 
     it "should fail if called directly from the parent class" do
-      expect { TestWorker.perform("message") }.to raise_error NotImplementedError
+      expect { TestConsumer.perform("message") }.to raise_error NotImplementedError
     end
   end
 
   describe "#options class" do
     it "should set options in class" do
-      expect(TestWorker.broker).to eq "aws"
+      expect(TestConsumer.options[:broker]).to eq "aws"
     end
   end
 
   describe "#build_instance_broker" do
     context "invalid provider" do
       before(:all) do
-        TestWorker.broker = "NaN"
+        TestConsumer.options[:broker] = "NaN"
       end
 
       after(:all) do
-        TestWorker.broker = "aws" # reverting
+        TestConsumer.options[:broker] = "aws" # reverting
       end
 
       it "should raise an error" do
-        expect { TestWorker.build_consumer_instance }.to raise_error PipefyMessage::Providers::Errors::InvalidOption
+        expect { TestConsumer.build_consumer_instance }.to raise_error PipefyMessage::Providers::Errors::InvalidOption
       end
     end