From 876a48a1abaa38b85495f55bb59a1b08679f931b Mon Sep 17 00:00:00 2001 From: Brusk Awat Date: Mon, 5 Aug 2024 00:54:02 +0300 Subject: [PATCH] Improves error handling and shut down. Moves code for reusability --- lib/rabbit_carrots.rb | 1 + lib/rabbit_carrots/core.rb | 113 ++++++++++++++++++++++++++++++ lib/rabbit_carrots/tasks/rmq.rake | 74 ++----------------- 3 files changed, 119 insertions(+), 69 deletions(-) create mode 100644 lib/rabbit_carrots/core.rb diff --git a/lib/rabbit_carrots.rb b/lib/rabbit_carrots.rb index 953d5dd..2f0a2c4 100644 --- a/lib/rabbit_carrots.rb +++ b/lib/rabbit_carrots.rb @@ -2,6 +2,7 @@ require_relative 'rabbit_carrots/version' require 'rabbit_carrots/connection' +require 'rabbit_carrots/core' require 'rabbit_carrots/configuration' require 'rabbit_carrots/railtie' if defined?(Rails) diff --git a/lib/rabbit_carrots/core.rb b/lib/rabbit_carrots/core.rb new file mode 100644 index 0000000..7846d45 --- /dev/null +++ b/lib/rabbit_carrots/core.rb @@ -0,0 +1,113 @@ +module RabbitCarrots + class Core + attr_reader :logger + + DatabaseAgonsticNotNullViolation = defined?(ActiveRecord) ? ActiveRecord::NotNullViolation : RabbitCarrots::EventHandlers::Errors::PlaceholderError + DatabaseAgonsticConnectionNotEstablished = defined?(ActiveRecord) ? ActiveRecord::ConnectionNotEstablished : Mongo::Error::SocketError + DatabaseAgnosticRecordInvalid = defined?(ActiveRecord) ? ActiveRecord::RecordInvalid : Mongoid::Errors::Validations + + def initialize(logger: nil) + @logger = logger || Logger.new(Rails.env.production? ? '/proc/self/fd/1' : $stdout) + @threads = [] + @running = true + @shutdown_requested = false + end + + def start(kill_to_restart_on_standard_error: false) + channels = RabbitCarrots.configuration.routing_key_mappings.map do |mapping| + { **mapping, handler: mapping[:handler].constantize } + end + + channels.each do |channel| + handler_class = channel[:handler] + raise "#{handler_class.name} must respond to `handle!`" unless handler_class.respond_to?(:handle!) + + @threads << Thread.new do + run_task( + queue_name: channel[:queue], + handler_class:, + routing_keys: channel[:routing_keys], + queue_arguments: channel[:arguments], + kill_to_restart_on_standard_error: + ) + end + end + + Signal.trap('INT') { request_shutdown } + Signal.trap('TERM') { request_shutdown } + + while @running + if @shutdown_requested + request_shutdown + sleep 1 + break + end + sleep 1 + end + + @threads.each(&:join) + rescue StandardError => e + logger.error "Error starting Rabbit Carrots: #{e.message}" + end + + def request_shutdown + # Workaround to a known issue with Signal Traps and logs + Thread.start do + logger.info 'Shutting down Rabbit Carrots service...' + end + @shutdown_requested = true + @threads.each(&:kill) + stop + end + + def stop + # Workaround to a known issue with Signal Traps and logs + Thread.start do + logger.info 'Stoppig the Rabbit Carrots service...' + end + @running = false + end + + def run_task(queue_name:, handler_class:, routing_keys:, queue_arguments: {}, kill_to_restart_on_standard_error: false) + RabbitCarrots::Connection.instance.channel.with do |channel| + exchange = channel.topic(RabbitCarrots.configuration.rabbitmq_exchange_name, durable: true) + + logger.info "Listening on QUEUE: #{queue_name} for ROUTING KEYS: #{routing_keys}" + queue = channel.queue(queue_name, durable: true, arguments: queue_arguments) + + routing_keys.map(&:strip).each { |k| queue.bind(exchange, routing_key: k) } + + queue.subscribe(block: false, manual_ack: true, prefetch: 10) do |delivery_info, properties, payload| + break if @shutdown_requested + + logger.info "Received from queue: #{queue_name}, Routing Keys: #{routing_keys}" + handler_class.handle!(channel, delivery_info, properties, payload) + channel.ack(delivery_info.delivery_tag, false) + rescue RabbitCarrots::EventHandlers::Errors::NackMessage, JSON::ParserError => _e + logger.info "Nacked message: #{payload}" + channel.nack(delivery_info.delivery_tag, false, false) + rescue RabbitCarrots::EventHandlers::Errors::NackAndRequeueMessage => _e + logger.info "Nacked and Requeued message: #{payload}" + channel.nack(delivery_info.delivery_tag, false, true) + rescue DatabaseAgonsticNotNullViolation, DatabaseAgnosticRecordInvalid => e + logger.info "Null constraint or Invalid violation: #{payload}. Error: #{e.message}" + channel.ack(delivery_info.delivery_tag, false) + rescue DatabaseAgonsticConnectionNotEstablished => e + logger.info "Error connection not established to the database: #{payload}. Error: #{e.message}" + sleep 3 + channel.nack(delivery_info.delivery_tag, false, true) + rescue StandardError => e + logger.info "Error handling message: #{payload}. Error: #{e.message}" + sleep 3 + channel.nack(delivery_info.delivery_tag, false, true) + Process.kill('SIGTERM', Process.pid) if kill_to_restart_on_standard_error + end + + logger.info "Ending task for queue: #{queue_name}" + end + rescue StandardError => e + logger.error "Bunny session error: #{e.message}" + request_shutdown + end + end +end diff --git a/lib/rabbit_carrots/tasks/rmq.rake b/lib/rabbit_carrots/tasks/rmq.rake index 541f45b..33efb5f 100644 --- a/lib/rabbit_carrots/tasks/rmq.rake +++ b/lib/rabbit_carrots/tasks/rmq.rake @@ -1,77 +1,13 @@ -require 'bunny' - namespace :rabbit_carrots do - desc 'Listener for Queue' + desc 'Rake task for standalone RabbitCarrots mode' task eat: :environment do Rails.application.eager_load! - # rubocop:disable Lint/ConstantDefinitionInBlock - DatabaseAgonsticNotNullViolation = defined?(ActiveRecord) ? ActiveRecord::NotNullViolation : RabbitCarrots::EventHandlers::Errors::PlaceholderError - DatabaseAgonsticConnectionNotEstablished = defined?(ActiveRecord) ? ActiveRecord::ConnectionNotEstablished : Mongo::Error::SocketError - DatabaseAgnosticRecordInvalid = defined?(ActiveRecord) ? ActiveRecord::RecordInvalid : Mongoid::Errors::Validations - # rubocop:enable Lint/ConstantDefinitionInBlock + logger = Logger.new(Rails.env.production? ? '/proc/self/fd/1' : $stdout) + logger.level = Logger::INFO - channels = RabbitCarrots.configuration.routing_key_mappings.map do |mapping| - # This will be supplied in initializer. At that time, the Handler will not be available to be loaded and will throw Uninitialized Constant - { **mapping, handler: mapping[:handler].constantize } - end + core_service = RabbitCarrots::Core.new(logger:) - Rails.logger = Logger.new(Rails.env.production? ? '/proc/self/fd/1' : $stdout) - - # Run RMQ Subscriber for each channel - channels.each do |channel| - handler_class = channel[:handler] - - raise "#{handler_class.name} must respond to `handle!`" unless handler_class.respond_to?(:handle!) - - run_task(queue_name: channel[:queue], handler_class:, routing_keys: channel[:routing_keys], queue_arguments: channel[:arguments]) - end - - # Infinite loop to keep the process running - loop do - sleep 1 - end - end -end - -def run_task(queue_name:, handler_class:, routing_keys:, queue_arguments: {}) - RabbitCarrots::Connection.instance.channel.with do |channel| - exchange = channel.topic(RabbitCarrots.configuration.rabbitmq_exchange_name, durable: true) - - Rails.logger.info "Listening on QUEUE: #{queue_name} for ROUTING KEYS: #{routing_keys}" - queue = channel.queue(queue_name, durable: true, arguments: queue_arguments) - - routing_keys.map(&:strip).each { |k| queue.bind(exchange, routing_key: k) } - - queue.subscribe(block: false, manual_ack: true, prefetch: 10) do |delivery_info, properties, payload| - Rails.logger.info "Received from queue: #{queue_name}, Routing Keys: #{routing_keys}" - handler_class.handle!(channel, delivery_info, properties, payload) - channel.ack(delivery_info.delivery_tag, false) - rescue RabbitCarrots::EventHandlers::Errors::NackMessage, JSON::ParserError => _e - Rails.logger.info "Nacked message: #{payload}" - channel.nack(delivery_info.delivery_tag, false, false) - rescue RabbitCarrots::EventHandlers::Errors::NackAndRequeueMessage => _e - Rails.logger.info "Nacked and Requeued message: #{payload}" - channel.nack(delivery_info.delivery_tag, false, true) - rescue DatabaseAgonsticNotNullViolation, DatabaseAgnosticRecordInvalid => e - # on null constraint violation, we want to ack the message - Rails.logger.error "Null constraint or Invalid violation: #{payload}. Error: #{e.message}" - channel.ack(delivery_info.delivery_tag, false) - rescue DatabaseAgonsticConnectionNotEstablished => e - # on connection not established, we want to requeue the message and sleep for 3 seconds - Rails.logger.error "Error connection not established to the database: #{payload}. Error: #{e.message}" - # delay for 3 seconds before requeuing - sleep 3 - channel.nack(delivery_info.delivery_tag, false, true) - rescue StandardError => e - Rails.logger.error "Error handling message: #{payload}. Error: #{e.message}" - # requeue the message then kill the container - sleep 3 - channel.nack(delivery_info.delivery_tag, false, true) - # kill the container with sigterm - Process.kill('SIGTERM', Process.pid) - end - - Rails.logger.info 'RUN TASK ENDED' + core_service.start(kill_to_restart_on_standard_error: true) end end