Class: Outboxable::PollingPublisherWorker
- Inherits:
-
Object
- Object
- Outboxable::PollingPublisherWorker
- Includes:
- Sidekiq::Job
- Defined in:
- lib/outboxable/polling_publisher_worker.rb
Instance Method Summary collapse
Instance Method Details
#perform(args) ⇒ Object
6 7 8 9 |
# File 'lib/outboxable/polling_publisher_worker.rb', line 6 def perform(args) orm = args['orm'] orm == 'mongoid' ? perform_mongoid(orm) : perform_activerecord(orm) end |
#perform_activerecord(orm) ⇒ Object
11 12 13 14 15 16 17 18 19 |
# File 'lib/outboxable/polling_publisher_worker.rb', line 11 def perform_activerecord(orm) 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) outbox.update(last_attempted_at: 1.minute.from_now, status: :processing, allow_publish: false) end end end |
#perform_mongoid(orm) ⇒ Object
21 22 23 24 25 26 27 |
# File 'lib/outboxable/polling_publisher_worker.rb', line 21 def perform_mongoid(orm) 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.idempotency_key, orm) outbox.update(last_attempted_at: 1.minute.from_now, status: :processing, allow_publish: false) end end |