diff --git a/lib/outboxable.rb b/lib/outboxable.rb index e7389b8..f1f8a3a 100644 --- a/lib/outboxable.rb +++ b/lib/outboxable.rb @@ -3,14 +3,15 @@ require_relative 'outboxable/version' require 'outboxable/worker' -require 'outboxable/publishing_manager' -require 'outboxable/polling_publisher_worker' require 'outboxable/connection' require 'outboxable/configuration' require 'outboxable/rabbitmq/publisher' require 'active_support/concern' +require 'outboxable/publishing_manager' +require 'outboxable/polling_publisher_worker' + module Outboxable class Error < StandardError; end diff --git a/lib/outboxable/configuration.rb b/lib/outboxable/configuration.rb index ea56822..c32c8dc 100644 --- a/lib/outboxable/configuration.rb +++ b/lib/outboxable/configuration.rb @@ -6,12 +6,6 @@ module Outboxable def self.configure self.configuration ||= Configuration.new yield(configuration) - - # In accordance to sidekiq-cron README: https://github.com/sidekiq-cron/sidekiq-cron#under-the-hood - Sidekiq::Options[:cron_poll_interval] = 5 - - # Create the cron job for the polling publisher - Sidekiq::Cron::Job.create(name: 'OutboxablePollingPublisher', cron: '*/5 * * * * *', class: 'Outboxable::PollingPublisherWorker', args: [{ orm: configuration.orm }]) end class Configuration @@ -32,6 +26,12 @@ module Outboxable raise Error, 'Outboxable Gem only supports Rails but you application does not seem to be a Rails app' unless Object.const_defined?('Rails') raise Error, 'Outboxable Gem only support Rails version 7 and newer' if Rails::VERSION::MAJOR < 7 raise Error, 'Outboxable Gem uses the sidekiq-cron Gem. Make sure you add it to your project' unless Object.const_defined?('Sidekiq::Cron') + + # In accordance to sidekiq-cron README: https://github.com/sidekiq-cron/sidekiq-cron#under-the-hood + Sidekiq::Options[:cron_poll_interval] = 5 + + # Create the cron job for the polling publisher + Sidekiq::Cron::Job.create(name: 'OutboxablePollingPublisher', cron: '*/5 * * * * *', class: 'Outboxable::PollingPublisherWorker') end def message_broker=(message_broker) diff --git a/lib/outboxable/polling_publisher_worker.rb b/lib/outboxable/polling_publisher_worker.rb index ecb1911..04b6cb3 100644 --- a/lib/outboxable/polling_publisher_worker.rb +++ b/lib/outboxable/polling_publisher_worker.rb @@ -1,27 +1,25 @@ module Outboxable class PollingPublisherWorker include Sidekiq::Job - sidekiq_options queue: 'critical' - def perform(args) - orm = args['orm'] - orm == 'mongoid' ? perform_mongoid(orm) : perform_activerecord(orm) + def perform + Outboxable.configuration.orm == :mongoid ? perform_mongoid : perform_activerecord end - def perform_activerecord(orm) + def perform_activerecord Outbox.pending.where(last_attempted_at: [..Time.zone.now, nil]).find_in_batches(batch_size: 100).each do |batch| batch.each do |outbox| # This is to prevent a job from being retried too many times. Worst-case scenario is 1 minute delay in jobs. - Outboxable::Worker.perform_async(outbox.id, orm) + ::Outboxable::Worker.perform_async(outbox.id) outbox.update(last_attempted_at: 1.minute.from_now, status: :processing, allow_publish: false) end end end - def perform_mongoid(orm) - Outbox.pending.where(last_attempted_at: [..Time.zone.now, nil]).each do |outbox| + def perform_mongoid + Outbox.pending.any_of({ last_attempted_at: ..Time.zone.now }, { last_attempted_at: nil }).each do |outbox| # This is to prevent a job from being retried too many times. Worst-case scenario is 1 minute delay in jobs. - Outboxable::Worker.perform_async(outbox.idempotency_key, orm) + ::Outboxable::Worker.perform_async(outbox.idempotency_key) outbox.update(last_attempted_at: 1.minute.from_now, status: :processing, allow_publish: false) end end diff --git a/lib/outboxable/worker.rb b/lib/outboxable/worker.rb index dc82d62..c699a61 100644 --- a/lib/outboxable/worker.rb +++ b/lib/outboxable/worker.rb @@ -4,9 +4,9 @@ module Outboxable class Worker include ::Sidekiq::Job - def perform(outbox_id, orm) - Outboxable::PublishingManager.publish(resource: Outbox.find(outbox_id)) if orm == 'activerecord' - Outboxable::PublishingManager.publish(resource: Outbox.find_by!(idempotency_key: outbox_id)) if orm == 'mongoid' + def perform(outbox_id) + Outboxable::PublishingManager.publish(resource: Outbox.find(outbox_id)) if Outboxable.configuration.orm == :activerecord + Outboxable::PublishingManager.publish(resource: Outbox.find_by!(idempotency_key: outbox_id)) if Outboxable.configuration.orm == :mongoid end end end