Class: Raises::Spool

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

Constant Summary collapse

RETRYABLE_STATUS = 
[408, 429].freeze

Instance Method Summary collapse

Constructor Details

#initialize(directory, deliver:, warn:, **options) ⇒ Spool

Returns a new instance of Spool.



9
10
11
12
13
14
15
16
17
 
# File 'lib/raises/spool.rb', line 9

def initialize(directory, deliver:, warn:, **options)
  @deliver = deliver
  @warn = warn
  @now = options.fetch(:now, -> { Time.now })
  @random = options.fetch(:random, Random.new)
  @start_on_enqueue = options.fetch(:start_on_enqueue, true)
  @storage = SpoolStorage.new(directory, warn: warn, now: @now)
  reset_process_state
end

Instance Method Details

#drain_once(limit: 20) ⇒ Object 



55
56
57
58
59
 
# File 'lib/raises/spool.rb', line 55

def drain_once(limit: 20)
  @storage.each_due(limit: limit) do |path, envelope|
    deliver(path, envelope)
  end
end

#enqueue(notice) ⇒ Object 



41
42
43
44
45
46
47
48
49
50
51
52
53
 
# File 'lib/raises/spool.rb', line 41

def enqueue(notice)
  accepted = @storage.store(notice) == :stored
  if accepted
    start if @start_on_enqueue
    wake if @start_on_enqueue
  else
    @warn.call("raises spool is full; notice was not queued")
  end
  accepted
rescue StandardError => e
  @warn.call("raises could not queue notice: #{e.class}: #{e.message}")
  false
end

#start ⇒ Object 



19
20
21
22
23
24
25
26
27
28
 
# File 'lib/raises/spool.rb', line 19

def start
  reset_process_state if @pid != Process.pid
  @mutex.synchronize do
    return if @thread&.alive?

    @thread = Thread.new { run }
    @thread.name = "raises-spool" if @thread.respond_to?(:name=)
    @thread.report_on_exception = false
  end
end

#stop ⇒ Object 



30
31
32
33
34
35
36
37
38
39
 
# File 'lib/raises/spool.rb', line 30

def stop
  return if @pid != Process.pid

  thread = @mutex.synchronize do
    @stopping = true
    @condition.broadcast
    @thread
  end
  thread&.join(1)
end