Adds support for puma

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

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}"