diff --git a/lib/outboxable/configuration.rb b/lib/outboxable/configuration.rb index 7bed8e8..c32c8dc 100644 --- a/lib/outboxable/configuration.rb +++ b/lib/outboxable/configuration.rb @@ -10,7 +10,7 @@ module Outboxable class Configuration ALLOWED_MESSAGE_BROKERS = %i[rabbitmq].freeze - ALLOWED_ORMS = %i[activerecord].freeze + ALLOWED_ORMS = %i[activerecord mongoid].freeze attr_accessor :rabbitmq_host, :rabbitmq_port, diff --git a/lib/outboxable/polling_publisher_worker.rb b/lib/outboxable/polling_publisher_worker.rb index 45757dc..c77c217 100644 --- a/lib/outboxable/polling_publisher_worker.rb +++ b/lib/outboxable/polling_publisher_worker.rb @@ -4,6 +4,10 @@ module Outboxable sidekiq_options queue: 'critical' def perform + Outboxable.configuration.orm == :mongoid ? perform_mongoid : perform_activerecord + end + + 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. @@ -12,5 +16,13 @@ module Outboxable end end end + + def perform_mongoid + Outbox.pending.where(last_attempted_at: [..Time.zone.now, 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.id) + outbox.update(last_attempted_at: 1.minute.from_now, status: :processing, allow_publish: false) + end + end end end