From a51eed8101e38b64fbccd93be3ca360d7dac530a Mon Sep 17 00:00:00 2001 From: Brusk Awat Date: Mon, 5 Aug 2024 01:03:10 +0300 Subject: [PATCH] Adds support for puma --- lib/puma/plugin/rabbit_carrots.rb | 84 +++++++++++++++++++++++++++++++ lib/rabbit_carrots/core.rb | 20 ++++---- 2 files changed, 94 insertions(+), 10 deletions(-) create mode 100644 lib/puma/plugin/rabbit_carrots.rb diff --git a/lib/puma/plugin/rabbit_carrots.rb b/lib/puma/plugin/rabbit_carrots.rb new file mode 100644 index 0000000..d49ce86 --- /dev/null +++ b/lib/puma/plugin/rabbit_carrots.rb @@ -0,0 +1,84 @@ +# rabbit_carrots.rb + +require 'puma/plugin' + +Puma::Plugin.create do + attr_reader :puma_pid, :rabbit_carrots_pid, :log_writer, :core_service + + def start(launcher) + @log_writer = launcher.log_writer + @puma_pid = $$ + + @core_service = RabbitCarrots::Core.new(logger: log_writer) + + in_background do + monitor_rabbit_carrots + end + + launcher.events.on_booted do + @rabbit_carrots_pid = fork do + Thread.new { monitor_puma } + start_rabbit_carrots_consumer + end + end + + launcher.events.on_stopped { stop_rabbit_carrots } + launcher.events.on_restart { stop_rabbit_carrots } + end + + private + + def start_rabbit_carrots_consumer + core_service.start(kill_to_restart_on_standard_error: true) + rescue StandardError => e + Rails.logger.error "Error starting Rabbit Carrots: #{e.message}" + end + + def stop_rabbit_carrots + return unless rabbit_carrots_pid + + log 'Stopping Rabbit Carrots...' + core_service.request_shutdown + Process.kill('TERM', rabbit_carrots_pid) + Process.wait(rabbit_carrots_pid) + rescue Errno::ECHILD, Errno::ESRCH + end + + def monitor_puma + monitor(:puma_dead?, 'Detected Puma has gone away, stopping Rabbit Carrots...') + end + + def monitor_rabbit_carrots + monitor(:rabbit_carrots_dead?, 'Rabbits Carrot is dead, stopping Puma...') + end + + def monitor(process_dead, message) + loop do + if send(process_dead) + log message + Process.kill('TERM', $$) + break + end + sleep 2 + end + end + + def rabbit_carrots_dead? + Process.waitpid(rabbit_carrots_pid, Process::WNOHANG) if rabbit_carrots_started? + false + rescue Errno::ECHILD, Errno::ESRCH + true + end + + def rabbit_carrots_started? + rabbit_carrots_pid.present? + end + + def puma_dead? + Process.ppid != puma_pid + end + + def log(...) + log_writer.log(...) + end +end diff --git a/lib/rabbit_carrots/core.rb b/lib/rabbit_carrots/core.rb index 7846d45..dba740d 100644 --- a/lib/rabbit_carrots/core.rb +++ b/lib/rabbit_carrots/core.rb @@ -53,7 +53,7 @@ module RabbitCarrots def request_shutdown # Workaround to a known issue with Signal Traps and logs Thread.start do - logger.info 'Shutting down Rabbit Carrots service...' + logger.log 'Shutting down Rabbit Carrots service...' end @shutdown_requested = true @threads.each(&:kill) @@ -63,7 +63,7 @@ module RabbitCarrots def stop # Workaround to a known issue with Signal Traps and logs Thread.start do - logger.info 'Stoppig the Rabbit Carrots service...' + logger.log 'Stoppig the Rabbit Carrots service...' end @running = false end @@ -72,7 +72,7 @@ module RabbitCarrots 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}" + logger.log "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) } @@ -80,30 +80,30 @@ module RabbitCarrots 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}" + logger.log "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}" + logger.log "Nacked message: #{payload}" channel.nack(delivery_info.delivery_tag, false, false) rescue RabbitCarrots::EventHandlers::Errors::NackAndRequeueMessage => _e - logger.info "Nacked and Requeued message: #{payload}" + logger.log "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}" + logger.log "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}" + logger.log "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}" + logger.log "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}" + logger.log "Ending task for queue: #{queue_name}" end rescue StandardError => e logger.error "Bunny session error: #{e.message}"