ditkrg/rabbit_carrots
4
0
Fork 0
mirror of https://github.com/ditkrg/rabbit_carrots.git synced 2026-01-22 22:06:40 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #6 from ditkrg/rewrite

Browse Source 
Rewrite
This commit is contained in:
Brusk Awat 2024-08-05 01:07:48 +03:00 committed by GitHub
commit edb1fffad1
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
7 changed files with 230 additions and 76 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
README.md
View File

@ -33,6 +33,9 @@ RabbitCarrots.configure do |c|
c.rabbitmq_password = ENV.fetch('RABBITMQ__PASSWORD', nil)
c.rabbitmq_vhost = ENV.fetch('RABBITMQ__VHOST', nil)
c.rabbitmq_exchange_name = ENV.fetch('RABBITMQ__EXCHANGE_NAME', nil)
c.automatically_recover = true
c.network_recovery_interval = 5
c.recovery_attempts = 5
c.routing_key_mappings = [
{ routing_keys: ['RK1', 'RK2'], queue: 'QUEUE_NAME', handler: 'CLASS HANDLER IN STRING' },
{ routing_keys: ['RK1', 'RK2'], queue: 'QUEUE_NAME', handler: 'CLASS HANDLER IN STRING' }
@ -41,8 +44,6 @@ end
```
Note that handler is a class that must implement a method named ```handle!``` that takes 4 parameters as follow:
```ruby
@ -53,8 +54,6 @@ class DummyEventHandler
end
```
Inside the handle message, you can NACK the message without re-queuing by raising ```RabbitCarrots::EventHandlers::Errors::NackMessage``` exception.
To NACK and re-queue, raise ```RabbitCarrots::EventHandlers::Errors::NackAndRequeueMessage``` exception.
@ -65,8 +64,16 @@ Note: Any other unrescued exception raised inside ```handle!``` the that is a su
### Running
Then run ```bundle exec rake rabbit_carrots:eat```.
For better scalability and improved performance, you can run rabbit_carrots in standalone mode by invoking the following command:
```bundle exec rake rabbit_carrots:eat```.
#### Puma
For small and medium sized projects, you can delegate the management of the rabbit_carrots to the Puma web server. To achieve that, add the following line to your puma.rb
```plugin :rabbit_carrots```
This will make sure that Puma will manage rabbit carrots as a background service and will gracefully terminate if rabbit_carrots eventually loses connection after multiple automatic recovery.
## Development
After checking out the repo, run `bin/setup` to install dependencies. Then, run `rake spec` to run the tests. You can also run `bin/console` for an interactive prompt that will allow you to experiment.

84
lib/puma/plugin/rabbit_carrots.rb Normal file
View File

@ -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

1
lib/rabbit_carrots.rb
View File

@ -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)

11
lib/rabbit_carrots/configuration.rb
View File

@ -9,6 +9,15 @@ module RabbitCarrots
end
class Configuration
attr_accessor :rabbitmq_host, :rabbitmq_port, :rabbitmq_user, :rabbitmq_password, :rabbitmq_vhost, :routing_key_mappings, :rabbitmq_exchange_name
attr_accessor :rabbitmq_host,
:rabbitmq_port,
:rabbitmq_user,
:rabbitmq_password,
:rabbitmq_vhost,
:routing_key_mappings,
:rabbitmq_exchange_name,
:automatically_recover,
:network_recovery_interval,
:recovery_attempts
end
end

6
lib/rabbit_carrots/connection.rb
View File

@ -11,7 +11,11 @@ module RabbitCarrots
port: RabbitCarrots.configuration.rabbitmq_port,
user: RabbitCarrots.configuration.rabbitmq_user,
password: RabbitCarrots.configuration.rabbitmq_password,
vhost: RabbitCarrots.configuration.rabbitmq_vhost
vhost: RabbitCarrots.configuration.rabbitmq_vhost,
automatically_recover: RabbitCarrots.configuration.automatically_recover || true,
network_recovery_interval: RabbitCarrots.configuration.network_recovery_interval || 5,
recovery_attempts: RabbitCarrots.configuration.recovery_attempts || 5,
recovery_attempts_exhausted: -> { Process.kill('TERM', Process.pid) }
)
@connection.start

113
lib/rabbit_carrots/core.rb Normal file
View File

@ -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.log '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.log '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.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) }
queue.subscribe(block: false, manual_ack: true, prefetch: 10) do |delivery_info, properties, payload|
break if @shutdown_requested
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.log "Nacked message: #{payload}"
channel.nack(delivery_info.delivery_tag, false, false)
rescue RabbitCarrots::EventHandlers::Errors::NackAndRequeueMessage => _e
logger.log "Nacked and Requeued message: #{payload}"
channel.nack(delivery_info.delivery_tag, false, true)
rescue DatabaseAgonsticNotNullViolation, DatabaseAgnosticRecordInvalid => e
logger.log "Null constraint or Invalid violation: #{payload}. Error: #{e.message}"
channel.ack(delivery_info.delivery_tag, false)
rescue DatabaseAgonsticConnectionNotEstablished => e
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.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.log "Ending task for queue: #{queue_name}"
end
rescue StandardError => e
logger.error "Bunny session error: #{e.message}"
request_shutdown
end
end
end

74
lib/rabbit_carrots/tasks/rmq.rake
View File

@ -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