Module: Postburner::Insertion

Extended by:
ActiveSupport::Concern
Included in:
Job
Defined in:
app/concerns/postburner/insertion.rb

Overview

Concern providing queue insertion methods for Postburner jobs.

Handles enqueuing jobs to Beanstalkd (or test strategies), including scheduling, re-queueing, and the internal insertion mechanics via after_save_commit callbacks.

Examples:

Basic queueing

class MyJob < Postburner::Job
  def perform(args)
    # ... work ...
  end
end

job = MyJob.create!(args: { foo: 'bar' })
job.queue!

Scheduled queueing

job.queue!(delay: 1.hour)
job.queue!(at: 2.days.from_now)

Instance Method Summary collapse

Instance Method Details

#queue!(options = {}) ⇒ true

Enqueues the job to Beanstalkd for processing.

Sets queued_at timestamp and optionally run_at for scheduled execution. Triggers enqueue callbacks and inserts job into Beanstalkd via after_save_commit hook. In test mode, executes immediately instead of queueing to Beanstalkd.

Examples:

Queue immediately

job.queue!

Queue with delay

job.queue!(delay: 1.hour)

Queue at specific time

job.queue!(at: Time.zone.now + 2.days)
job.queue!(at: Time.zone.parse('2025-01-15 09:00:00'))
job.queue!(at: '2025-01-15 09:00:00'.in_time_zone)
job.queue!(at: Time.parse('2025-01-15 09:00:00 EST'))

Queue with priority and TTR

job.queue!(priority: 0, ttr: 600)

Queue to specific queue

job.queue!(queue: 'critical', delay: 30.minutes)

Parameters:

  • options (Hash) (defaults to: {}) —

    Queue options

Options Hash (options):

  • :at (Time, ActiveSupport::Duration) —

    Absolute time to run the job

  • :delay (Integer, ActiveSupport::Duration) —

    Seconds to delay execution

  • :priority (Integer) —

    Priority (0-4294967295, 0 = HIGHEST), sets instance attribute

  • :pri (Integer) —

    Beanstalkd priority (pass-through, for backwards compatibility)

  • :ttr (Integer) —

    Time-to-run in seconds (1-4294967295, 0 is silently changed to 1)

  • :queue (String) —

    Queue name override

Returns:

  • (true) —

    on success (including if already queued)

Raises:

  • (ActiveRecord::RecordInvalid) —

    if job is not valid

  • (AlreadyProcessed) —

    if job was already processed

See Also:



71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
 
# File 'app/concerns/postburner/insertion.rb', line 71

def queue!(options={})
  return true if self.queued_at.present? && self.bkid.present?
  raise ActiveRecord::RecordInvalid, "Can't queue unless valid." unless self.valid?
  raise AlreadyProcessed, "Processed at #{self.processed_at}" if self.processed_at

  # Extract and set instance-level overrides
  self.priority = options.delete(:priority) if options.key?(:priority)
  self.ttr = options.delete(:ttr) if options.key?(:ttr)
  self.queue_name = options.delete(:queue) if options.key?(:queue)

  at = options.delete(:at)
  now = Time.current

  self.queued_at = now
  self.run_at = case
                when at.present?
                  # this is rudimentary, add error handling
                  #
                  # Ceil, don't floor. Beanstalkd delays are whole seconds,
                  # so some rounding is unavoidable -- but it must round
                  # *away* from now. Flooring both sides (`at.to_i -
                  # now.to_i`) drops frac(at) and frac(now) independently,
                  # releasing the job early by frac(at) - frac(now)
                  # whenever that is positive. Consumers then see a job
                  # arrive before its own run_at.
                  options[:delay] ||= [(at.to_time - now).ceil, 0].max
                  at
                when options[:delay].present?
                  now + options[:delay].seconds
                end

  @_insert_options = options

  run_callbacks :enqueue do
    self.save!
  end

  true
end

#requeue!(options = {}) ⇒ true

Re-queues an existing job by removing it from Beanstalkd and queueing again.

Calls #delete! to remove from Beanstalkd, resets queuing metadata, then calls #queue! with new options.

Examples:

Requeue with different delay

job.requeue!(delay: 5.minutes)

Parameters:

  • options (Hash) (defaults to: {}) —

    Queue options (same as #queue!)

Options Hash (options):

  • :at (Time, ActiveSupport::Duration) —

    Absolute time to run the job

  • :delay (Integer, ActiveSupport::Duration) —

    Seconds to delay execution

  • :priority (Integer) —

    Priority (0-4294967295, lower = higher priority)

  • :ttr (Integer) —

    Time-to-run in seconds (1-4294967295)

  • :queue (String) —

    Queue name override

Returns:

  • (true) —

    on success

Raises:

  • (ActiveRecord::RecordInvalid) —

    if job is not valid

  • (Beaneater::NotConnected) —

    if Beanstalkd connection fails

See Also:



134
135
136
137
138
139
 
# File 'app/concerns/postburner/insertion.rb', line 134

def requeue!(options={})
  self.delete!
  self.bkid, self.queued_at = nil, nil

  self.queue! options
end

#will_insert? ⇒ Boolean

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Checks if job is flagged for insertion into Beanstalkd.

Set internally by #queue! to trigger insertion via after_save_commit hook.

Returns:

  • (Boolean) —

    true if job will be inserted on save



148
149
150
 
# File 'app/concerns/postburner/insertion.rb', line 148

def will_insert?
  @_insert_options.is_a? Hash
end