Class: Honker::Outbox

Inherits:
Object
  • Object
show all
Defined in:
lib/honker.rb

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(db, name, delivery, visibility_timeout_s:, max_attempts:, base_backoff_s:) ⇒ Outbox

Returns a new instance of Outbox.

Raises:

  • (ArgumentError) 


474
475
476
477
478
479
480
481
482
483
484
485
486
 
# File 'lib/honker.rb', line 474

def initialize(db, name, delivery, visibility_timeout_s:, max_attempts:, base_backoff_s:)
  raise ArgumentError, "delivery must respond to #call" unless delivery.respond_to?(:call)

  @name = name
  @delivery = delivery
  @max_attempts = max_attempts
  @base_backoff_s = base_backoff_s
  @queue = db.queue(
    "_outbox:#{name}",
    visibility_timeout_s: visibility_timeout_s,
    max_attempts: max_attempts,
  )
end

Instance Attribute Details

#base_backoff_s ⇒ Object (readonly)

Returns the value of attribute base_backoff_s.



472
473
474
 
# File 'lib/honker.rb', line 472

def base_backoff_s
  @base_backoff_s
end

#max_attempts ⇒ Object (readonly)

Returns the value of attribute max_attempts.



472
473
474
 
# File 'lib/honker.rb', line 472

def max_attempts
  @max_attempts
end

#name ⇒ Object (readonly)

Returns the value of attribute name.



472
473
474
 
# File 'lib/honker.rb', line 472

def name
  @name
end

#queue ⇒ Object (readonly)

Returns the value of attribute queue.



472
473
474
 
# File 'lib/honker.rb', line 472

def queue
  @queue
end

Instance Method Details

#enqueue(payload, tx: nil, delay: nil, run_at: nil, priority: 0, expires: nil) ⇒ Object 



488
489
490
491
492
493
494
 
# File 'lib/honker.rb', line 488

def enqueue(payload, tx: nil, delay: nil, run_at: nil, priority: 0, expires: nil)
  if tx
    @queue.enqueue_tx(tx, payload, delay: delay, run_at: run_at, priority: priority, expires: expires)
  else
    @queue.enqueue(payload, delay: delay, run_at: run_at, priority: priority, expires: expires)
  end
end

#run_once(worker_id) ⇒ Object 



496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
 
# File 'lib/honker.rb', line 496

def run_once(worker_id)
  job = @queue.claim_one(worker_id)
  return false unless job

  begin
    if @delivery.arity == 1
      @delivery.call(job.payload)
    else
      @delivery.call(job.payload, job)
    end
    raise "outbox ack failed for job #{job.id}" unless job.ack
  rescue StandardError => e
    delay_s = retry_delay(job.attempts)
    raise "outbox retry failed for job #{job.id}" unless job.retry(delay_s: delay_s, error: "#{e}\n#{e.backtrace&.join("\n")}")
  end
  true
end

#run_worker(worker_id, stop: nil, idle_sleep_s: 0.1) ⇒ Object 



514
515
516
517
518
519
 
# File 'lib/honker.rb', line 514

def run_worker(worker_id, stop: nil, idle_sleep_s: 0.1)
  until stop&.call
    processed = run_once(worker_id)
    sleep(idle_sleep_s) unless processed
  end
end